// Copyright 2022 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package v2

import (
	"encoding/json"
	"time"

	"github.com/pingcap/ticdc/pkg/api"
	bf "github.com/pingcap/ticdc/pkg/binlog-filter"
	"github.com/pingcap/ticdc/pkg/common"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/integrity"
	"github.com/pingcap/ticdc/pkg/security"
	"github.com/pingcap/ticdc/pkg/util"
)

// EmptyResponse return empty {} to http client
type EmptyResponse struct{}

// LogLevelReq log level request
type LogLevelReq struct {
	Level string `json:"log_level"`
}

// RedactModeReq redaction mode request
// Use redact_info_log to match CLI flag naming convention
// Accepts: "off", "on", "marker" (case-insensitive)
type RedactModeReq struct {
	Mode string `json:"redact_info_log"`
}

// RedactModeResp redaction mode response
type RedactModeResp struct {
	PreviousMode string `json:"previous_mode"`
	CurrentMode  string `json:"current_mode"`
}

// ListResponse is the response for all List APIs
type ListResponse[T any] struct {
	Total int `json:"total"`
	Items []T `json:"items"`
}

// Tso contains timestamp get from PD
type Tso struct {
	Timestamp int64 `json:"timestamp"`
	LogicTime int64 `json:"logic_time"`
}

// Tables contains IneligibleTables, EligibleTables and AllTables
type Tables struct {
	IneligibleTables []TableName `json:"ineligible_tables,omitempty"`
	EligibleTables   []TableName `json:"eligible_tables,omitempty"`
	AllTables        []TableName `json:"all_tables,omitempty"`
}

// TableName contains table information
type TableName struct {
	Schema      string `json:"database_name"`
	Table       string `json:"table_name"`
	TableID     int64  `json:"table_id"`
	IsPartition bool   `json:"is_partition"`
}

// VerifyTableConfig use to verify tables.
// Only use by Open API v2.
type VerifyTableConfig struct {
	PDConfig
	ReplicaConfig *ReplicaConfig `json:"replica_config"`
	StartTs       uint64         `json:"start_ts"`
	SinkURI       string         `json:"sink_uri"`
}

func getDefaultVerifyTableConfig() *VerifyTableConfig {
	return &VerifyTableConfig{
		ReplicaConfig: GetDefaultReplicaConfig(),
	}
}

// ResumeChangefeedConfig is used by resume changefeed api
type ResumeChangefeedConfig struct {
	PDConfig
	OverwriteCheckpointTs uint64 `json:"overwrite_checkpoint_ts"`
}

// PDConfig is a configuration used to connect to pd
type PDConfig struct {
	PDAddrs       []string `json:"pd_addrs,omitempty"`
	CAPath        string   `json:"ca_path"`
	CertPath      string   `json:"cert_path"`
	KeyPath       string   `json:"key_path"`
	CertAllowedCN []string `json:"cert_allowed_cn,omitempty"`
}

// ChangefeedCommonInfo holds some common usage information of a changefeed
type ChangefeedCommonInfo struct {
	UpstreamID     uint64               `json:"upstream_id"`
	ID             string               `json:"id"`
	Keyspace       string               `json:"keyspace"`
	FeedState      config.FeedState     `json:"state"`
	CheckpointTSO  uint64               `json:"checkpoint_tso"`
	CheckpointTime api.JSONTime         `json:"checkpoint_time"`
	RunningError   *config.RunningError `json:"error"`
}

// SyncedStatusConfig represents synced check interval config for a changefeed
type SyncedStatusConfig struct {
	// The minimum interval between the latest synced ts and now required to reach synced state
	SyncedCheckInterval *int64 `json:"synced_check_interval"`
	// The maximum interval between latest checkpoint ts and now or
	// between latest sink's checkpoint ts and puller's checkpoint ts required to reach synced state
	CheckpointInterval *int64 `json:"checkpoint_interval"`
}

// MarshalJSON marshal changefeed common info to json
// we need to set feed state to normal if it is uninitialized and pending to warning
// to hide the detail of uninitialized and pending state from user
func (c ChangefeedCommonInfo) MarshalJSON() ([]byte, error) {
	// alias the original type to prevent recursive call of MarshalJSON
	type Alias ChangefeedCommonInfo

	if c.FeedState == config.StateUnInitialized {
		c.FeedState = config.StateNormal
	}
	if c.FeedState == config.StatePending {
		c.FeedState = config.StateWarning
	}

	return json.Marshal(struct {
		Alias
	}{
		Alias: Alias(c),
	})
}

// ChangefeedConfig use by create changefeed api
type ChangefeedConfig struct {
	Keyspace      string         `json:"keyspace"`
	ID            string         `json:"changefeed_id"`
	StartTs       uint64         `json:"start_ts"`
	TargetTs      uint64         `json:"target_ts"`
	SinkURI       string         `json:"sink_uri"`
	ReplicaConfig *ReplicaConfig `json:"replica_config"`
	PDConfig
}

// ProcessorCommonInfo holds the common info of a processor
type ProcessorCommonInfo struct {
	Keyspace     string `json:"keyspace"`
	ChangeFeedID string `json:"changefeed_id"`
	CaptureID    string `json:"capture_id"`
}

// JSONDuration used to wrap duration into json format
type JSONDuration struct {
	duration time.Duration
}

// MarshalJSON marshal duration to string
func (d JSONDuration) MarshalJSON() ([]byte, error) {
	return json.Marshal(d.duration.Nanoseconds())
}

// UnmarshalJSON unmarshal json value to wrapped duration
func (d *JSONDuration) UnmarshalJSON(b []byte) error {
	var v interface{}
	if err := json.Unmarshal(b, &v); err != nil {
		return err
	}
	switch value := v.(type) {
	case float64:
		d.duration = time.Duration(value)
		return nil
	case string:
		var err error
		d.duration, err = time.ParseDuration(value)
		if err != nil {
			return err
		}
		return nil
	default:
		return errors.New("invalid duration")
	}
}

// ReplicaConfig is a duplicate of  config.ReplicaConfig
type ReplicaConfig struct {
	MemoryQuota           *uint64 `json:"memory_quota,omitempty"`
	CaseSensitive         *bool   `json:"case_sensitive,omitempty"`
	ForceReplicate        *bool   `json:"force_replicate,omitempty"`
	IgnoreIneligibleTable *bool   `json:"ignore_ineligible_table,omitempty"`
	CheckGCSafePoint      *bool   `json:"check_gc_safe_point,omitempty"`
	EnableSyncPoint       *bool   `json:"enable_sync_point,omitempty"`
	EnableTableMonitor    *bool   `json:"enable_table_monitor,omitempty"`
	BDRMode               *bool   `json:"bdr_mode,omitempty"`
	// EnableActiveActive enables active-active replication mode on top of BDR.
	// It requires BDRMode to be true and is only supported by TiDB and storage sinks.
	EnableActiveActive *bool `json:"enable_active_active,omitempty"`
	// ActiveActiveProgressInterval controls how often the MySQL/TiDB sink updates the
	// active-active progress table in EnableActiveActive mode (for hard delete safety checks).
	ActiveActiveProgressInterval *JSONDuration `json:"active_active_progress_interval,omitempty"`
	// ActiveActiveSyncStatsInterval controls how often the MySQL/TiDB sink queries
	// the TiDB session variable @@tidb_cdc_active_active_sync_stats for conflict statistics.
	// Set it to 0 to disable metric collection.
	// This option only takes effect when EnableActiveActive is true and the downstream is TiDB.
	ActiveActiveSyncStatsInterval *JSONDuration `json:"active_active_sync_stats_interval,omitempty"`

	SyncPointInterval  *JSONDuration `json:"sync_point_interval,omitempty"`
	SyncPointRetention *JSONDuration `json:"sync_point_retention,omitempty"`

	Filter                       *FilterConfig              `json:"filter,omitempty"`
	Mounter                      *MounterConfig             `json:"mounter,omitempty"`
	Sink                         *SinkConfig                `json:"sink,omitempty"`
	Consistent                   *ConsistentConfig          `json:"consistent,omitempty"`
	Scheduler                    *ChangefeedSchedulerConfig `json:"scheduler,omitempty"`
	Integrity                    *IntegrityConfig           `json:"integrity,omitempty"`
	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
	SyncedStatus                 *SyncedStatusConfig        `json:"synced_status,omitempty"`

	// Deprecated: we don't use this field since v8.0.0.
	SQLMode *string `json:"sql_mode,omitempty"`
}

// ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
func (c *ReplicaConfig) ToInternalReplicaConfig() *config.ReplicaConfig {
	return c.toInternalReplicaConfigWithOriginConfig(config.GetDefaultReplicaConfig())
}

// ToInternalReplicaConfigWithOriginConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
	res *config.ReplicaConfig,
) *config.ReplicaConfig {
	if c.MemoryQuota != nil {
		res.MemoryQuota = c.MemoryQuota
	}
	if c.CaseSensitive != nil {
		res.CaseSensitive = c.CaseSensitive
	}
	if c.ForceReplicate != nil {
		res.ForceReplicate = c.ForceReplicate
	}
	if c.CheckGCSafePoint != nil {
		res.CheckGCSafePoint = c.CheckGCSafePoint
	}
	res.EnableSyncPoint = c.EnableSyncPoint
	res.EnableTableMonitor = c.EnableTableMonitor
	if c.IgnoreIneligibleTable != nil {
		res.IgnoreIneligibleTable = c.IgnoreIneligibleTable
	}
	if c.SyncPointInterval != nil {
		res.SyncPointInterval = &c.SyncPointInterval.duration
	}
	if c.SyncPointRetention != nil {
		res.SyncPointRetention = &c.SyncPointRetention.duration
	}
	if c.BDRMode != nil {
		res.BDRMode = c.BDRMode
	}
	if c.EnableActiveActive != nil {
		res.EnableActiveActive = c.EnableActiveActive
	}
	if c.ActiveActiveProgressInterval != nil {
		res.ActiveActiveProgressInterval = &c.ActiveActiveProgressInterval.duration
	}
	if c.ActiveActiveSyncStatsInterval != nil {
		res.ActiveActiveSyncStatsInterval = &c.ActiveActiveSyncStatsInterval.duration
	}

	if c.Filter != nil {
		efs := make([]*config.EventFilterRule, 0, len(c.Filter.EventFilters))
		for _, ef := range c.Filter.EventFilters {
			efs = append(efs, ef.ToInternalEventFilterRule())
		}
		res.Filter = &config.FilterConfig{
			Rules:            c.Filter.Rules,
			IgnoreTxnStartTs: c.Filter.IgnoreTxnStartTs,
			EventFilters:     efs,
		}
	}
	if c.Consistent != nil {
		if res.Consistent == nil {
			res.Consistent = &config.ConsistentConfig{}
		}

		if c.Consistent.Level != nil {
			res.Consistent.Level = c.Consistent.Level
		}
		if c.Consistent.MaxLogSize != nil {
			res.Consistent.MaxLogSize = c.Consistent.MaxLogSize
		}
		if c.Consistent.FlushIntervalInMs != nil {
			res.Consistent.FlushIntervalInMs = c.Consistent.FlushIntervalInMs
		}
		if c.Consistent.MetaFlushIntervalInMs != nil {
			res.Consistent.MetaFlushIntervalInMs = c.Consistent.MetaFlushIntervalInMs
		}
		if c.Consistent.EncodingWorkerNum != nil {
			res.Consistent.EncodingWorkerNum = c.Consistent.EncodingWorkerNum
		}
		if c.Consistent.FlushWorkerNum != nil {
			res.Consistent.FlushWorkerNum = c.Consistent.FlushWorkerNum
		}
		if c.Consistent.Storage != nil {
			res.Consistent.Storage = c.Consistent.Storage
		}
		if c.Consistent.UseFileBackend != nil {
			res.Consistent.UseFileBackend = c.Consistent.UseFileBackend
		}
		if c.Consistent.Compression != nil {
			res.Consistent.Compression = c.Consistent.Compression
		}
		if c.Consistent.FlushConcurrency != nil {
			res.Consistent.FlushConcurrency = c.Consistent.FlushConcurrency
		}

		if c.Consistent.MemoryUsage != nil {
			res.Consistent.MemoryUsage = &config.ConsistentMemoryUsage{
				MemoryQuotaPercentage: c.Consistent.MemoryUsage.MemoryQuotaPercentage,
			}
		}
	}
	if c.Sink != nil {
		var dispatchRules []*config.DispatchRule
		for _, rule := range c.Sink.DispatchRules {
			dispatchRules = append(dispatchRules, &config.DispatchRule{
				Matcher:        rule.Matcher,
				DispatcherRule: "",
				PartitionRule:  rule.PartitionRule,
				IndexName:      rule.IndexName,
				Columns:        rule.Columns,
				TopicRule:      rule.TopicRule,
			})
		}
		var columnSelectors []*config.ColumnSelector
		for _, selector := range c.Sink.ColumnSelectors {
			columnSelectors = append(columnSelectors, &config.ColumnSelector{
				Matcher: selector.Matcher,
				Columns: selector.Columns,
			})
		}
		var csvConfig *config.CSVConfig
		if c.Sink.CSVConfig != nil {
			csvConfig = &config.CSVConfig{
				Delimiter:            c.Sink.CSVConfig.Delimiter,
				Quote:                c.Sink.CSVConfig.Quote,
				NullString:           c.Sink.CSVConfig.NullString,
				IncludeCommitTs:      c.Sink.CSVConfig.IncludeCommitTs,
				BinaryEncodingMethod: c.Sink.CSVConfig.BinaryEncodingMethod,
				OutputOldValue:       c.Sink.CSVConfig.OutputOldValue,
				OutputHandleKey:      c.Sink.CSVConfig.OutputHandleKey,
				OutputFieldHeader:    c.Sink.CSVConfig.OutputFieldHeader,
			}
		}
		var pulsarConfig *config.PulsarConfig
		if c.Sink.PulsarConfig != nil {
			pulsarConfig = &config.PulsarConfig{
				TLSKeyFilePath:          c.Sink.PulsarConfig.TLSKeyFilePath,
				TLSCertificateFile:      c.Sink.PulsarConfig.TLSCertificateFile,
				TLSTrustCertsFilePath:   c.Sink.PulsarConfig.TLSTrustCertsFilePath,
				PulsarProducerCacheSize: c.Sink.PulsarConfig.PulsarProducerCacheSize,
				PulsarVersion:           c.Sink.PulsarConfig.PulsarVersion,
				CompressionType:         (*config.PulsarCompressionType)(c.Sink.PulsarConfig.CompressionType),
				AuthenticationToken:     c.Sink.PulsarConfig.AuthenticationToken,
				ConnectionTimeout:       (*config.TimeSec)(c.Sink.PulsarConfig.ConnectionTimeout),
				OperationTimeout:        (*config.TimeSec)(c.Sink.PulsarConfig.OperationTimeout),
				BatchingMaxMessages:     c.Sink.PulsarConfig.BatchingMaxMessages,
				BatchingMaxPublishDelay: (*config.TimeMill)(c.Sink.PulsarConfig.BatchingMaxPublishDelay),
				SendTimeout:             (*config.TimeSec)(c.Sink.PulsarConfig.SendTimeout),
				TokenFromFile:           c.Sink.PulsarConfig.TokenFromFile,
				BasicUserName:           c.Sink.PulsarConfig.BasicUserName,
				BasicPassword:           c.Sink.PulsarConfig.BasicPassword,
				AuthTLSCertificatePath:  c.Sink.PulsarConfig.AuthTLSCertificatePath,
				AuthTLSPrivateKeyPath:   c.Sink.PulsarConfig.AuthTLSPrivateKeyPath,
				OutputRawChangeEvent:    c.Sink.PulsarConfig.OutputRawChangeEvent,
			}
			if c.Sink.PulsarConfig.OAuth2 != nil {
				pulsarConfig.OAuth2 = &config.OAuth2{
					OAuth2IssuerURL:  c.Sink.PulsarConfig.OAuth2.OAuth2IssuerURL,
					OAuth2Audience:   c.Sink.PulsarConfig.OAuth2.OAuth2Audience,
					OAuth2PrivateKey: c.Sink.PulsarConfig.OAuth2.OAuth2PrivateKey,
					OAuth2ClientID:   c.Sink.PulsarConfig.OAuth2.OAuth2ClientID,
					OAuth2Scope:      c.Sink.PulsarConfig.OAuth2.OAuth2Scope,
				}
			}
		}

		var kafkaConfig *config.KafkaConfig
		if c.Sink.KafkaConfig != nil {
			var codeConfig *config.CodecConfig
			if c.Sink.KafkaConfig.CodecConfig != nil {
				oldConfig := c.Sink.KafkaConfig.CodecConfig
				codeConfig = &config.CodecConfig{
					EnableTiDBExtension:            oldConfig.EnableTiDBExtension,
					MaxBatchSize:                   oldConfig.MaxBatchSize,
					AvroEnableWatermark:            oldConfig.AvroEnableWatermark,
					AvroDecimalHandlingMode:        oldConfig.AvroDecimalHandlingMode,
					AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode,
					EncodingFormat:                 oldConfig.EncodingFormat,
				}
			}

			var largeMessageHandle *config.LargeMessageHandleConfig
			if c.Sink.KafkaConfig.LargeMessageHandle != nil {
				oldConfig := c.Sink.KafkaConfig.LargeMessageHandle
				largeMessageHandle = &config.LargeMessageHandleConfig{
					LargeMessageHandleOption:      oldConfig.LargeMessageHandleOption,
					LargeMessageHandleCompression: oldConfig.LargeMessageHandleCompression,
					ClaimCheckStorageURI:          oldConfig.ClaimCheckStorageURI,
					ClaimCheckRawValue:            oldConfig.ClaimCheckRawValue,
				}
			}

			var glueSchemaRegistryConfig *config.GlueSchemaRegistryConfig
			if c.Sink.KafkaConfig.GlueSchemaRegistryConfig != nil {
				glueSchemaRegistryConfig = &config.GlueSchemaRegistryConfig{
					RegistryName:    c.Sink.KafkaConfig.GlueSchemaRegistryConfig.RegistryName,
					Region:          c.Sink.KafkaConfig.GlueSchemaRegistryConfig.Region,
					AccessKey:       c.Sink.KafkaConfig.GlueSchemaRegistryConfig.AccessKey,
					SecretAccessKey: c.Sink.KafkaConfig.GlueSchemaRegistryConfig.SecretAccessKey,
					Token:           c.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token,
				}
			}

			kafkaConfig = &config.KafkaConfig{
				PartitionNum:                 c.Sink.KafkaConfig.PartitionNum,
				ReplicationFactor:            c.Sink.KafkaConfig.ReplicationFactor,
				KafkaVersion:                 c.Sink.KafkaConfig.KafkaVersion,
				MaxMessageBytes:              c.Sink.KafkaConfig.MaxMessageBytes,
				Compression:                  c.Sink.KafkaConfig.Compression,
				KafkaClientID:                c.Sink.KafkaConfig.KafkaClientID,
				AutoCreateTopic:              c.Sink.KafkaConfig.AutoCreateTopic,
				DialTimeout:                  c.Sink.KafkaConfig.DialTimeout,
				WriteTimeout:                 c.Sink.KafkaConfig.WriteTimeout,
				ReadTimeout:                  c.Sink.KafkaConfig.ReadTimeout,
				RequiredAcks:                 c.Sink.KafkaConfig.RequiredAcks,
				SASLUser:                     c.Sink.KafkaConfig.SASLUser,
				SASLPassword:                 c.Sink.KafkaConfig.SASLPassword,
				SASLMechanism:                c.Sink.KafkaConfig.SASLMechanism,
				SASLGssAPIAuthType:           c.Sink.KafkaConfig.SASLGssAPIAuthType,
				SASLGssAPIKeytabPath:         c.Sink.KafkaConfig.SASLGssAPIKeytabPath,
				SASLGssAPIKerberosConfigPath: c.Sink.KafkaConfig.SASLGssAPIKerberosConfigPath,
				SASLGssAPIServiceName:        c.Sink.KafkaConfig.SASLGssAPIServiceName,
				SASLGssAPIUser:               c.Sink.KafkaConfig.SASLGssAPIUser,
				SASLGssAPIPassword:           c.Sink.KafkaConfig.SASLGssAPIPassword,
				SASLGssAPIRealm:              c.Sink.KafkaConfig.SASLGssAPIRealm,
				SASLGssAPIDisablePafxfast:    c.Sink.KafkaConfig.SASLGssAPIDisablePafxfast,
				SASLOAuthClientID:            c.Sink.KafkaConfig.SASLOAuthClientID,
				SASLOAuthClientSecret:        c.Sink.KafkaConfig.SASLOAuthClientSecret,
				SASLOAuthTokenURL:            c.Sink.KafkaConfig.SASLOAuthTokenURL,
				SASLOAuthScopes:              c.Sink.KafkaConfig.SASLOAuthScopes,
				SASLOAuthGrantType:           c.Sink.KafkaConfig.SASLOAuthGrantType,
				SASLOAuthAudience:            c.Sink.KafkaConfig.SASLOAuthAudience,
				EnableTLS:                    c.Sink.KafkaConfig.EnableTLS,
				CA:                           c.Sink.KafkaConfig.CA,
				Cert:                         c.Sink.KafkaConfig.Cert,
				Key:                          c.Sink.KafkaConfig.Key,
				InsecureSkipVerify:           c.Sink.KafkaConfig.InsecureSkipVerify,
				CodecConfig:                  codeConfig,
				LargeMessageHandle:           largeMessageHandle,
				GlueSchemaRegistryConfig:     glueSchemaRegistryConfig,
				OutputRawChangeEvent:         c.Sink.KafkaConfig.OutputRawChangeEvent,
			}
		}
		var mysqlConfig *config.MySQLConfig
		if c.Sink.MySQLConfig != nil {
			mysqlConfig = &config.MySQLConfig{
				WorkerCount:                  c.Sink.MySQLConfig.WorkerCount,
				MaxTxnRow:                    c.Sink.MySQLConfig.MaxTxnRow,
				MaxMultiUpdateRowSize:        c.Sink.MySQLConfig.MaxMultiUpdateRowSize,
				MaxMultiUpdateRowCount:       c.Sink.MySQLConfig.MaxMultiUpdateRowCount,
				TiDBTxnMode:                  c.Sink.MySQLConfig.TiDBTxnMode,
				SSLCa:                        c.Sink.MySQLConfig.SSLCa,
				SSLCert:                      c.Sink.MySQLConfig.SSLCert,
				SSLKey:                       c.Sink.MySQLConfig.SSLKey,
				TimeZone:                     c.Sink.MySQLConfig.TimeZone,
				WriteTimeout:                 c.Sink.MySQLConfig.WriteTimeout,
				ReadTimeout:                  c.Sink.MySQLConfig.ReadTimeout,
				Timeout:                      c.Sink.MySQLConfig.Timeout,
				EnableBatchDML:               c.Sink.MySQLConfig.EnableBatchDML,
				EnableMultiStatement:         c.Sink.MySQLConfig.EnableMultiStatement,
				EnableCachePreparedStatement: c.Sink.MySQLConfig.EnableCachePreparedStatement,
			}
		}
		var cloudStorageConfig *config.CloudStorageConfig
		if c.Sink.CloudStorageConfig != nil {
			cloudStorageConfig = &config.CloudStorageConfig{
				WorkerCount:                  c.Sink.CloudStorageConfig.WorkerCount,
				FlushInterval:                c.Sink.CloudStorageConfig.FlushInterval,
				FileSize:                     c.Sink.CloudStorageConfig.FileSize,
				OutputColumnID:               c.Sink.CloudStorageConfig.OutputColumnID,
				FileExpirationDays:           c.Sink.CloudStorageConfig.FileExpirationDays,
				FileCleanupCronSpec:          c.Sink.CloudStorageConfig.FileCleanupCronSpec,
				FlushConcurrency:             c.Sink.CloudStorageConfig.FlushConcurrency,
				EnableSchemaIndexByGetObject: c.Sink.CloudStorageConfig.EnableSchemaIndexByGetObject,
				OutputRawChangeEvent:         c.Sink.CloudStorageConfig.OutputRawChangeEvent,
			}
		}
		var debeziumConfig *config.DebeziumConfig
		if c.Sink.DebeziumConfig != nil {
			debeziumConfig = &config.DebeziumConfig{
				OutputOldValue: c.Sink.DebeziumConfig.OutputOldValue,
			}
		}
		var openProtocolConfig *config.OpenProtocolConfig
		if c.Sink.OpenProtocolConfig != nil {
			openProtocolConfig = &config.OpenProtocolConfig{
				OutputOldValue: c.Sink.OpenProtocolConfig.OutputOldValue,
			}
		}

		res.Sink = &config.SinkConfig{
			DispatchRules:                    dispatchRules,
			Protocol:                         c.Sink.Protocol,
			CSVConfig:                        csvConfig,
			ColumnSelectors:                  columnSelectors,
			SchemaRegistry:                   c.Sink.SchemaRegistry,
			EncoderConcurrency:               c.Sink.EncoderConcurrency,
			Terminator:                       c.Sink.Terminator,
			DateSeparator:                    c.Sink.DateSeparator,
			EnablePartitionSeparator:         c.Sink.EnablePartitionSeparator,
			FileIndexWidth:                   c.Sink.FileIndexWidth,
			OnlyOutputUpdatedColumns:         c.Sink.OnlyOutputUpdatedColumns,
			DeleteOnlyOutputHandleKeyColumns: c.Sink.DeleteOnlyOutputHandleKeyColumns,
			ContentCompatible:                c.Sink.ContentCompatible,
			OnlyOutputPKColumns:              c.Sink.OnlyOutputPKColumns,
			KafkaConfig:                      kafkaConfig,
			MySQLConfig:                      mysqlConfig,
			PulsarConfig:                     pulsarConfig,
			CloudStorageConfig:               cloudStorageConfig,
			SafeMode:                         c.Sink.SafeMode,
			OpenProtocol:                     openProtocolConfig,
			Debezium:                         debeziumConfig,
		}

		if c.Sink.TxnAtomicity != nil {
			res.Sink.TxnAtomicity = util.AddressOf(config.AtomicityLevel(*c.Sink.TxnAtomicity))
		}
		if c.Sink.AdvanceTimeoutInSec != nil {
			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*c.Sink.AdvanceTimeoutInSec)
		}
		if c.Sink.DebeziumDisableSchema != nil {
			res.Sink.DebeziumDisableSchema = util.AddressOf(*c.Sink.DebeziumDisableSchema)
		}

		if c.Sink.SendBootstrapIntervalInSec != nil {
			res.Sink.SendBootstrapIntervalInSec = util.AddressOf(*c.Sink.SendBootstrapIntervalInSec)
		}

		if c.Sink.SendBootstrapInMsgCount != nil {
			res.Sink.SendBootstrapInMsgCount = util.AddressOf(*c.Sink.SendBootstrapInMsgCount)
		}

		if c.Sink.SendBootstrapToAllPartition != nil {
			res.Sink.SendBootstrapToAllPartition = util.AddressOf(*c.Sink.SendBootstrapToAllPartition)
		}

		if c.Sink.SendAllBootstrapAtStart != nil {
			res.Sink.SendAllBootstrapAtStart = util.AddressOf(*c.Sink.SendAllBootstrapAtStart)
		}
	}
	if c.Mounter != nil {
		if res.Mounter == nil {
			res.Mounter = &config.MounterConfig{}
		}

		if c.Mounter.WorkerNum != nil {
			res.Mounter.WorkerNum = *c.Mounter.WorkerNum
		}
	}
	if c.Scheduler != nil {
		if res.Scheduler == nil {
			res.Scheduler = &config.ChangefeedSchedulerConfig{}
		}

		if c.Scheduler.EnableTableAcrossNodes != nil {
			res.Scheduler.EnableTableAcrossNodes = c.Scheduler.EnableTableAcrossNodes
		}
		if c.Scheduler.RegionThreshold != nil {
			res.Scheduler.RegionThreshold = c.Scheduler.RegionThreshold
		}
		if c.Scheduler.RegionCountPerSpan != nil {
			res.Scheduler.RegionCountPerSpan = c.Scheduler.RegionCountPerSpan
		}
		if c.Scheduler.RegionCountRefreshInterval != nil {
			res.Scheduler.RegionCountRefreshInterval = c.Scheduler.RegionCountRefreshInterval
		}
		if c.Scheduler.WriteKeyThreshold != nil {
			res.Scheduler.WriteKeyThreshold = c.Scheduler.WriteKeyThreshold
		}
		if c.Scheduler.SchedulingTaskCountPerNode != nil {
			res.Scheduler.SchedulingTaskCountPerNode = c.Scheduler.SchedulingTaskCountPerNode
		}
		if c.Scheduler.EnableSplittableCheck != nil {
			res.Scheduler.EnableSplittableCheck = c.Scheduler.EnableSplittableCheck
		}
		if c.Scheduler.ForceSplit != nil {
			res.Scheduler.ForceSplit = c.Scheduler.ForceSplit
		}

		if c.Scheduler.BalanceScoreThreshold != nil {
			res.Scheduler.BalanceScoreThreshold = c.Scheduler.BalanceScoreThreshold
		}
		if c.Scheduler.MinTrafficPercentage != nil {
			res.Scheduler.MinTrafficPercentage = c.Scheduler.MinTrafficPercentage
		}
		if c.Scheduler.MaxTrafficPercentage != nil {
			res.Scheduler.MaxTrafficPercentage = c.Scheduler.MaxTrafficPercentage
		}
	}
	if c.Integrity != nil {
		if res.Integrity == nil {
			res.Integrity = &integrity.Config{}
		}

		if c.Integrity.IntegrityCheckLevel != nil {
			res.Integrity.IntegrityCheckLevel = c.Integrity.IntegrityCheckLevel
		}
		if c.Integrity.CorruptionHandleLevel != nil {
			res.Integrity.CorruptionHandleLevel = c.Integrity.CorruptionHandleLevel
		}
	}
	if c.ChangefeedErrorStuckDuration != nil {
		res.ChangefeedErrorStuckDuration = &c.ChangefeedErrorStuckDuration.duration
	}
	if c.SyncedStatus != nil {
		if res.SyncedStatus == nil {
			res.SyncedStatus = &config.SyncedStatusConfig{}
		}

		if c.SyncedStatus.SyncedCheckInterval != nil {
			res.SyncedStatus.SyncedCheckInterval = c.SyncedStatus.SyncedCheckInterval
		}
		if c.SyncedStatus.CheckpointInterval != nil {
			res.SyncedStatus.CheckpointInterval = c.SyncedStatus.CheckpointInterval
		}
	}
	return res
}

// ToAPIReplicaConfig coverts *config.ReplicaConfig into *v2.ReplicaConfig
func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
	cloned := c.Clone()

	res := &ReplicaConfig{
		MemoryQuota:           cloned.MemoryQuota,
		CaseSensitive:         cloned.CaseSensitive,
		ForceReplicate:        cloned.ForceReplicate,
		IgnoreIneligibleTable: cloned.IgnoreIneligibleTable,
		CheckGCSafePoint:      cloned.CheckGCSafePoint,
		EnableSyncPoint:       cloned.EnableSyncPoint,
		EnableTableMonitor:    cloned.EnableTableMonitor,
		BDRMode:               cloned.BDRMode,
		EnableActiveActive:    cloned.EnableActiveActive,
	}

	if cloned.SyncPointInterval != nil {
		res.SyncPointInterval = &JSONDuration{*cloned.SyncPointInterval}
	}

	if cloned.SyncPointRetention != nil {
		res.SyncPointRetention = &JSONDuration{*cloned.SyncPointRetention}
	}
	if cloned.ActiveActiveProgressInterval != nil {
		res.ActiveActiveProgressInterval = &JSONDuration{*cloned.ActiveActiveProgressInterval}
	}
	if cloned.ActiveActiveSyncStatsInterval != nil {
		res.ActiveActiveSyncStatsInterval = &JSONDuration{*cloned.ActiveActiveSyncStatsInterval}
	}

	if cloned.Filter != nil {
		efs := make([]EventFilterRule, 0, len(cloned.Filter.EventFilters))
		for _, ef := range cloned.Filter.EventFilters {
			efs = append(efs, ToAPIEventFilterRule(ef))
		}

		res.Filter = &FilterConfig{
			Rules:            cloned.Filter.Rules,
			IgnoreTxnStartTs: cloned.Filter.IgnoreTxnStartTs,
			EventFilters:     efs,
		}
	}
	if cloned.Sink != nil {
		var dispatchRules []*DispatchRule
		for _, rule := range cloned.Sink.DispatchRules {
			dispatchRules = append(dispatchRules, &DispatchRule{
				Matcher:       rule.Matcher,
				PartitionRule: rule.PartitionRule,
				IndexName:     rule.IndexName,
				Columns:       rule.Columns,
				TopicRule:     rule.TopicRule,
			})
		}
		var columnSelectors []*ColumnSelector
		for _, selector := range cloned.Sink.ColumnSelectors {
			columnSelectors = append(columnSelectors, &ColumnSelector{
				Matcher: selector.Matcher,
				Columns: selector.Columns,
			})
		}
		var csvConfig *CSVConfig
		if cloned.Sink.CSVConfig != nil {
			csvConfig = &CSVConfig{
				Delimiter:            cloned.Sink.CSVConfig.Delimiter,
				Quote:                cloned.Sink.CSVConfig.Quote,
				NullString:           cloned.Sink.CSVConfig.NullString,
				IncludeCommitTs:      cloned.Sink.CSVConfig.IncludeCommitTs,
				BinaryEncodingMethod: cloned.Sink.CSVConfig.BinaryEncodingMethod,
				OutputOldValue:       cloned.Sink.CSVConfig.OutputOldValue,
				OutputHandleKey:      cloned.Sink.CSVConfig.OutputHandleKey,
				OutputFieldHeader:    cloned.Sink.CSVConfig.OutputFieldHeader,
			}
		}
		var kafkaConfig *KafkaConfig
		if cloned.Sink.KafkaConfig != nil {
			var codeConfig *CodecConfig
			if cloned.Sink.KafkaConfig.CodecConfig != nil {
				oldConfig := cloned.Sink.KafkaConfig.CodecConfig
				codeConfig = &CodecConfig{
					EnableTiDBExtension:            oldConfig.EnableTiDBExtension,
					MaxBatchSize:                   oldConfig.MaxBatchSize,
					AvroEnableWatermark:            oldConfig.AvroEnableWatermark,
					AvroDecimalHandlingMode:        oldConfig.AvroDecimalHandlingMode,
					AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode,
					EncodingFormat:                 oldConfig.EncodingFormat,
				}
			}

			var largeMessageHandle *LargeMessageHandleConfig
			if cloned.Sink.KafkaConfig.LargeMessageHandle != nil {
				oldConfig := cloned.Sink.KafkaConfig.LargeMessageHandle
				largeMessageHandle = &LargeMessageHandleConfig{
					LargeMessageHandleOption:      oldConfig.LargeMessageHandleOption,
					LargeMessageHandleCompression: oldConfig.LargeMessageHandleCompression,
					ClaimCheckStorageURI:          oldConfig.ClaimCheckStorageURI,
					ClaimCheckRawValue:            oldConfig.ClaimCheckRawValue,
				}
			}

			var glueSchemaRegistryConfig *GlueSchemaRegistryConfig
			if cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig != nil {
				glueSchemaRegistryConfig = &GlueSchemaRegistryConfig{
					RegistryName:    cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.RegistryName,
					Region:          cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.Region,
					AccessKey:       cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.AccessKey,
					SecretAccessKey: cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.SecretAccessKey,
					Token:           cloned.Sink.KafkaConfig.GlueSchemaRegistryConfig.Token,
				}
			}

			kafkaConfig = &KafkaConfig{
				PartitionNum:                 cloned.Sink.KafkaConfig.PartitionNum,
				ReplicationFactor:            cloned.Sink.KafkaConfig.ReplicationFactor,
				KafkaVersion:                 cloned.Sink.KafkaConfig.KafkaVersion,
				MaxMessageBytes:              cloned.Sink.KafkaConfig.MaxMessageBytes,
				Compression:                  cloned.Sink.KafkaConfig.Compression,
				KafkaClientID:                cloned.Sink.KafkaConfig.KafkaClientID,
				AutoCreateTopic:              cloned.Sink.KafkaConfig.AutoCreateTopic,
				DialTimeout:                  cloned.Sink.KafkaConfig.DialTimeout,
				WriteTimeout:                 cloned.Sink.KafkaConfig.WriteTimeout,
				ReadTimeout:                  cloned.Sink.KafkaConfig.ReadTimeout,
				RequiredAcks:                 cloned.Sink.KafkaConfig.RequiredAcks,
				SASLUser:                     cloned.Sink.KafkaConfig.SASLUser,
				SASLPassword:                 cloned.Sink.KafkaConfig.SASLPassword,
				SASLMechanism:                cloned.Sink.KafkaConfig.SASLMechanism,
				SASLGssAPIAuthType:           cloned.Sink.KafkaConfig.SASLGssAPIAuthType,
				SASLGssAPIKeytabPath:         cloned.Sink.KafkaConfig.SASLGssAPIKeytabPath,
				SASLGssAPIKerberosConfigPath: cloned.Sink.KafkaConfig.SASLGssAPIKerberosConfigPath,
				SASLGssAPIServiceName:        cloned.Sink.KafkaConfig.SASLGssAPIServiceName,
				SASLGssAPIUser:               cloned.Sink.KafkaConfig.SASLGssAPIUser,
				SASLGssAPIPassword:           cloned.Sink.KafkaConfig.SASLGssAPIPassword,
				SASLGssAPIRealm:              cloned.Sink.KafkaConfig.SASLGssAPIRealm,
				SASLGssAPIDisablePafxfast:    cloned.Sink.KafkaConfig.SASLGssAPIDisablePafxfast,
				SASLOAuthClientID:            cloned.Sink.KafkaConfig.SASLOAuthClientID,
				SASLOAuthClientSecret:        cloned.Sink.KafkaConfig.SASLOAuthClientSecret,
				SASLOAuthTokenURL:            cloned.Sink.KafkaConfig.SASLOAuthTokenURL,
				SASLOAuthScopes:              cloned.Sink.KafkaConfig.SASLOAuthScopes,
				SASLOAuthGrantType:           cloned.Sink.KafkaConfig.SASLOAuthGrantType,
				SASLOAuthAudience:            cloned.Sink.KafkaConfig.SASLOAuthAudience,
				EnableTLS:                    cloned.Sink.KafkaConfig.EnableTLS,
				CA:                           cloned.Sink.KafkaConfig.CA,
				Cert:                         cloned.Sink.KafkaConfig.Cert,
				Key:                          cloned.Sink.KafkaConfig.Key,
				InsecureSkipVerify:           cloned.Sink.KafkaConfig.InsecureSkipVerify,
				CodecConfig:                  codeConfig,
				LargeMessageHandle:           largeMessageHandle,
				GlueSchemaRegistryConfig:     glueSchemaRegistryConfig,
				OutputRawChangeEvent:         cloned.Sink.KafkaConfig.OutputRawChangeEvent,
			}
		}
		var mysqlConfig *MySQLConfig
		if cloned.Sink.MySQLConfig != nil {
			mysqlConfig = &MySQLConfig{
				WorkerCount:                  cloned.Sink.MySQLConfig.WorkerCount,
				MaxTxnRow:                    cloned.Sink.MySQLConfig.MaxTxnRow,
				MaxMultiUpdateRowSize:        cloned.Sink.MySQLConfig.MaxMultiUpdateRowSize,
				MaxMultiUpdateRowCount:       cloned.Sink.MySQLConfig.MaxMultiUpdateRowCount,
				TiDBTxnMode:                  cloned.Sink.MySQLConfig.TiDBTxnMode,
				SSLCa:                        cloned.Sink.MySQLConfig.SSLCa,
				SSLCert:                      cloned.Sink.MySQLConfig.SSLCert,
				SSLKey:                       cloned.Sink.MySQLConfig.SSLKey,
				TimeZone:                     cloned.Sink.MySQLConfig.TimeZone,
				WriteTimeout:                 cloned.Sink.MySQLConfig.WriteTimeout,
				ReadTimeout:                  cloned.Sink.MySQLConfig.ReadTimeout,
				Timeout:                      cloned.Sink.MySQLConfig.Timeout,
				EnableBatchDML:               cloned.Sink.MySQLConfig.EnableBatchDML,
				EnableMultiStatement:         cloned.Sink.MySQLConfig.EnableMultiStatement,
				EnableCachePreparedStatement: cloned.Sink.MySQLConfig.EnableCachePreparedStatement,
			}
		}
		var pulsarConfig *PulsarConfig
		if cloned.Sink.PulsarConfig != nil {
			pulsarConfig = &PulsarConfig{
				TLSKeyFilePath:          cloned.Sink.PulsarConfig.TLSKeyFilePath,
				TLSCertificateFile:      cloned.Sink.PulsarConfig.TLSCertificateFile,
				TLSTrustCertsFilePath:   cloned.Sink.PulsarConfig.TLSTrustCertsFilePath,
				PulsarProducerCacheSize: cloned.Sink.PulsarConfig.PulsarProducerCacheSize,
				PulsarVersion:           cloned.Sink.PulsarConfig.PulsarVersion,
				CompressionType:         (*string)(cloned.Sink.PulsarConfig.CompressionType),
				AuthenticationToken:     cloned.Sink.PulsarConfig.AuthenticationToken,
				ConnectionTimeout:       (*int)(cloned.Sink.PulsarConfig.ConnectionTimeout),
				OperationTimeout:        (*int)(cloned.Sink.PulsarConfig.OperationTimeout),
				BatchingMaxMessages:     cloned.Sink.PulsarConfig.BatchingMaxMessages,
				BatchingMaxPublishDelay: (*int)(cloned.Sink.PulsarConfig.BatchingMaxPublishDelay),
				SendTimeout:             (*int)(cloned.Sink.PulsarConfig.SendTimeout),
				TokenFromFile:           cloned.Sink.PulsarConfig.TokenFromFile,
				BasicUserName:           cloned.Sink.PulsarConfig.BasicUserName,
				BasicPassword:           cloned.Sink.PulsarConfig.BasicPassword,
				AuthTLSCertificatePath:  cloned.Sink.PulsarConfig.AuthTLSCertificatePath,
				AuthTLSPrivateKeyPath:   cloned.Sink.PulsarConfig.AuthTLSPrivateKeyPath,
				OutputRawChangeEvent:    cloned.Sink.PulsarConfig.OutputRawChangeEvent,
			}
			if cloned.Sink.PulsarConfig.OAuth2 != nil {
				pulsarConfig.OAuth2 = &PulsarOAuth2{
					OAuth2IssuerURL:  cloned.Sink.PulsarConfig.OAuth2.OAuth2IssuerURL,
					OAuth2Audience:   cloned.Sink.PulsarConfig.OAuth2.OAuth2Audience,
					OAuth2PrivateKey: cloned.Sink.PulsarConfig.OAuth2.OAuth2PrivateKey,
					OAuth2ClientID:   cloned.Sink.PulsarConfig.OAuth2.OAuth2ClientID,
					OAuth2Scope:      cloned.Sink.PulsarConfig.OAuth2.OAuth2Scope,
				}
			}
		}
		var cloudStorageConfig *CloudStorageConfig
		if cloned.Sink.CloudStorageConfig != nil {
			cloudStorageConfig = &CloudStorageConfig{
				WorkerCount:                  cloned.Sink.CloudStorageConfig.WorkerCount,
				FlushInterval:                cloned.Sink.CloudStorageConfig.FlushInterval,
				FileSize:                     cloned.Sink.CloudStorageConfig.FileSize,
				OutputColumnID:               cloned.Sink.CloudStorageConfig.OutputColumnID,
				FileExpirationDays:           cloned.Sink.CloudStorageConfig.FileExpirationDays,
				FileCleanupCronSpec:          cloned.Sink.CloudStorageConfig.FileCleanupCronSpec,
				FlushConcurrency:             cloned.Sink.CloudStorageConfig.FlushConcurrency,
				EnableSchemaIndexByGetObject: cloned.Sink.CloudStorageConfig.EnableSchemaIndexByGetObject,
				OutputRawChangeEvent:         cloned.Sink.CloudStorageConfig.OutputRawChangeEvent,
			}
		}
		var debeziumConfig *DebeziumConfig
		if cloned.Sink.Debezium != nil {
			debeziumConfig = &DebeziumConfig{
				OutputOldValue: cloned.Sink.Debezium.OutputOldValue,
			}
		}
		var openProtocolConfig *OpenProtocolConfig
		if cloned.Sink.OpenProtocol != nil {
			openProtocolConfig = &OpenProtocolConfig{
				OutputOldValue: cloned.Sink.OpenProtocol.OutputOldValue,
			}
		}
		res.Sink = &SinkConfig{
			Protocol:                         cloned.Sink.Protocol,
			SchemaRegistry:                   cloned.Sink.SchemaRegistry,
			DispatchRules:                    dispatchRules,
			CSVConfig:                        csvConfig,
			ColumnSelectors:                  columnSelectors,
			EncoderConcurrency:               cloned.Sink.EncoderConcurrency,
			Terminator:                       cloned.Sink.Terminator,
			DateSeparator:                    cloned.Sink.DateSeparator,
			EnablePartitionSeparator:         cloned.Sink.EnablePartitionSeparator,
			FileIndexWidth:                   cloned.Sink.FileIndexWidth,
			OnlyOutputUpdatedColumns:         cloned.Sink.OnlyOutputUpdatedColumns,
			DeleteOnlyOutputHandleKeyColumns: cloned.Sink.DeleteOnlyOutputHandleKeyColumns,
			ContentCompatible:                cloned.Sink.ContentCompatible,
			OnlyOutputPKColumns:              cloned.Sink.OnlyOutputPKColumns,
			KafkaConfig:                      kafkaConfig,
			MySQLConfig:                      mysqlConfig,
			PulsarConfig:                     pulsarConfig,
			CloudStorageConfig:               cloudStorageConfig,
			SafeMode:                         cloned.Sink.SafeMode,
			DebeziumConfig:                   debeziumConfig,
			OpenProtocolConfig:               openProtocolConfig,
		}

		if cloned.Sink.TxnAtomicity != nil {
			res.Sink.TxnAtomicity = util.AddressOf(string(*cloned.Sink.TxnAtomicity))
		}
		if cloned.Sink.AdvanceTimeoutInSec != nil {
			res.Sink.AdvanceTimeoutInSec = util.AddressOf(*cloned.Sink.AdvanceTimeoutInSec)
		}

		if cloned.Sink.SendBootstrapIntervalInSec != nil {
			res.Sink.SendBootstrapIntervalInSec = util.AddressOf(*cloned.Sink.SendBootstrapIntervalInSec)
		}

		if cloned.Sink.SendBootstrapInMsgCount != nil {
			res.Sink.SendBootstrapInMsgCount = util.AddressOf(*cloned.Sink.SendBootstrapInMsgCount)
		}

		if cloned.Sink.SendBootstrapToAllPartition != nil {
			res.Sink.SendBootstrapToAllPartition = util.AddressOf(*cloned.Sink.SendBootstrapToAllPartition)
		}

		if cloned.Sink.SendAllBootstrapAtStart != nil {
			res.Sink.SendAllBootstrapAtStart = util.AddressOf(*cloned.Sink.SendAllBootstrapAtStart)
		}

		if cloned.Sink.DebeziumDisableSchema != nil {
			res.Sink.DebeziumDisableSchema = util.AddressOf(*cloned.Sink.DebeziumDisableSchema)
		}
	}
	if cloned.Consistent != nil {
		if res.Consistent == nil {
			res.Consistent = &ConsistentConfig{}
		}

		if cloned.Consistent.Level != nil {
			res.Consistent.Level = cloned.Consistent.Level
		}
		if cloned.Consistent.MaxLogSize != nil {
			res.Consistent.MaxLogSize = cloned.Consistent.MaxLogSize
		}
		if cloned.Consistent.FlushIntervalInMs != nil {
			res.Consistent.FlushIntervalInMs = cloned.Consistent.FlushIntervalInMs
		}
		if cloned.Consistent.MetaFlushIntervalInMs != nil {
			res.Consistent.MetaFlushIntervalInMs = cloned.Consistent.MetaFlushIntervalInMs
		}
		if cloned.Consistent.EncodingWorkerNum != nil {
			res.Consistent.EncodingWorkerNum = cloned.Consistent.EncodingWorkerNum
		}
		if cloned.Consistent.FlushWorkerNum != nil {
			res.Consistent.FlushWorkerNum = cloned.Consistent.FlushWorkerNum
		}
		if cloned.Consistent.Storage != nil {
			res.Consistent.Storage = cloned.Consistent.Storage
		}
		if cloned.Consistent.UseFileBackend != nil {
			res.Consistent.UseFileBackend = cloned.Consistent.UseFileBackend
		}
		if cloned.Consistent.Compression != nil {
			res.Consistent.Compression = cloned.Consistent.Compression
		}
		if cloned.Consistent.FlushConcurrency != nil {
			res.Consistent.FlushConcurrency = cloned.Consistent.FlushConcurrency
		}
		if cloned.Consistent.MemoryUsage != nil {
			res.Consistent.MemoryUsage = &ConsistentMemoryUsage{
				MemoryQuotaPercentage: cloned.Consistent.MemoryUsage.MemoryQuotaPercentage,
			}
		}
	}

	if cloned.Mounter != nil {
		res.Mounter = &MounterConfig{
			WorkerNum: &cloned.Mounter.WorkerNum,
		}
	}
	if cloned.Scheduler != nil {
		if res.Scheduler == nil {
			res.Scheduler = &ChangefeedSchedulerConfig{}
		}

		if cloned.Scheduler.EnableTableAcrossNodes != nil {
			res.Scheduler.EnableTableAcrossNodes = cloned.Scheduler.EnableTableAcrossNodes
		}
		if cloned.Scheduler.RegionThreshold != nil {
			res.Scheduler.RegionThreshold = cloned.Scheduler.RegionThreshold
		}
		if cloned.Scheduler.RegionCountPerSpan != nil {
			res.Scheduler.RegionCountPerSpan = cloned.Scheduler.RegionCountPerSpan
		}
		if cloned.Scheduler.RegionCountRefreshInterval != nil {
			res.Scheduler.RegionCountRefreshInterval = cloned.Scheduler.RegionCountRefreshInterval
		}
		if cloned.Scheduler.WriteKeyThreshold != nil {
			res.Scheduler.WriteKeyThreshold = cloned.Scheduler.WriteKeyThreshold
		}
		if cloned.Scheduler.SchedulingTaskCountPerNode != nil {
			res.Scheduler.SchedulingTaskCountPerNode = cloned.Scheduler.SchedulingTaskCountPerNode
		}
		if cloned.Scheduler.EnableSplittableCheck != nil {
			res.Scheduler.EnableSplittableCheck = cloned.Scheduler.EnableSplittableCheck
		}
		if cloned.Scheduler.ForceSplit != nil {
			res.Scheduler.ForceSplit = cloned.Scheduler.ForceSplit
		}
		if cloned.Scheduler.BalanceScoreThreshold != nil {
			res.Scheduler.BalanceScoreThreshold = cloned.Scheduler.BalanceScoreThreshold
		}
		if cloned.Scheduler.MinTrafficPercentage != nil {
			res.Scheduler.MinTrafficPercentage = cloned.Scheduler.MinTrafficPercentage
		}
		if cloned.Scheduler.MaxTrafficPercentage != nil {
			res.Scheduler.MaxTrafficPercentage = cloned.Scheduler.MaxTrafficPercentage
		}
	}

	if cloned.Integrity != nil {
		if res.Integrity == nil {
			res.Integrity = &IntegrityConfig{}
		}

		if cloned.Integrity.IntegrityCheckLevel != nil {
			res.Integrity.IntegrityCheckLevel = cloned.Integrity.IntegrityCheckLevel
		}
		if cloned.Integrity.CorruptionHandleLevel != nil {
			res.Integrity.CorruptionHandleLevel = cloned.Integrity.CorruptionHandleLevel
		}
	}
	if cloned.ChangefeedErrorStuckDuration != nil {
		res.ChangefeedErrorStuckDuration = &JSONDuration{*cloned.ChangefeedErrorStuckDuration}
	}
	if cloned.SyncedStatus != nil {
		if res.SyncedStatus == nil {
			res.SyncedStatus = &SyncedStatusConfig{}
		}
		if cloned.SyncedStatus.SyncedCheckInterval != nil {
			res.SyncedStatus.SyncedCheckInterval = cloned.SyncedStatus.SyncedCheckInterval
		}
		if cloned.SyncedStatus.CheckpointInterval != nil {
			res.SyncedStatus.CheckpointInterval = cloned.SyncedStatus.CheckpointInterval
		}
	}
	return res
}

// GetDefaultReplicaConfig returns a default ReplicaConfig
func GetDefaultReplicaConfig() *ReplicaConfig {
	return ToAPIReplicaConfig(config.GetDefaultReplicaConfig())
}

// FilterConfig represents filter config for a changefeed
// This is a duplicate of config.FilterConfig
type FilterConfig struct {
	Rules            []string          `json:"rules,omitempty"`
	IgnoreTxnStartTs []uint64          `json:"ignore_txn_start_ts,omitempty"`
	EventFilters     []EventFilterRule `json:"event_filters,omitempty"`
}

// MounterConfig represents mounter config for a changefeed
type MounterConfig struct {
	WorkerNum *int `json:"worker_num,omitempty"`
}

// EventFilterRule is used by sql event filter and expression filter
type EventFilterRule struct {
	Matcher     []string `json:"matcher"`
	IgnoreEvent []string `json:"ignore_event"`
	// regular expression
	IgnoreSQL []string `toml:"ignore_sql" json:"ignore_sql"`
	// sql expression
	IgnoreInsertValueExpr    string `json:"ignore_insert_value_expr"`
	IgnoreUpdateNewValueExpr string `json:"ignore_update_new_value_expr"`
	IgnoreUpdateOldValueExpr string `json:"ignore_update_old_value_expr"`
	IgnoreDeleteValueExpr    string `json:"ignore_delete_value_expr"`
}

// ToInternalEventFilterRule converts EventFilterRule to *config.EventFilterRule
func (e EventFilterRule) ToInternalEventFilterRule() *config.EventFilterRule {
	res := &config.EventFilterRule{
		Matcher:                  e.Matcher,
		IgnoreSQL:                e.IgnoreSQL,
		IgnoreInsertValueExpr:    &e.IgnoreInsertValueExpr,
		IgnoreUpdateNewValueExpr: &e.IgnoreUpdateNewValueExpr,
		IgnoreUpdateOldValueExpr: &e.IgnoreUpdateOldValueExpr,
		IgnoreDeleteValueExpr:    &e.IgnoreDeleteValueExpr,
	}
	if len(e.IgnoreEvent) != 0 {
		res.IgnoreEvent = make([]bf.EventType, len(e.IgnoreEvent))
		for i, et := range e.IgnoreEvent {
			res.IgnoreEvent[i] = bf.EventType(et)
		}
	}
	return res
}

// ToAPIEventFilterRule converts *config.EventFilterRule to API EventFilterRule
func ToAPIEventFilterRule(er *config.EventFilterRule) EventFilterRule {
	res := EventFilterRule{
		IgnoreInsertValueExpr:    util.GetOrZero(er.IgnoreInsertValueExpr),
		IgnoreUpdateNewValueExpr: util.GetOrZero(er.IgnoreUpdateNewValueExpr),
		IgnoreUpdateOldValueExpr: util.GetOrZero(er.IgnoreUpdateOldValueExpr),
		IgnoreDeleteValueExpr:    util.GetOrZero(er.IgnoreDeleteValueExpr),
	}
	if len(er.Matcher) != 0 {
		res.Matcher = make([]string, len(er.Matcher))
		copy(res.Matcher, er.Matcher)
	}
	if len(er.IgnoreSQL) != 0 {
		res.IgnoreSQL = make([]string, len(er.IgnoreSQL))
		copy(res.IgnoreSQL, er.IgnoreSQL)
	}
	if len(er.IgnoreEvent) != 0 {
		res.IgnoreEvent = make([]string, len(er.IgnoreEvent))
		for i, et := range er.IgnoreEvent {
			res.IgnoreEvent[i] = string(et)
		}
	}
	return res
}

// Table represents a qualified table name.
type Table struct {
	// Schema is the name of the schema (database) containing this table.
	Schema string `json:"database_name"`
	// Name is the unqualified table name.
	Name string `json:"table_name"`
}

// SinkConfig represents sink config for a changefeed
// This is a duplicate of config.SinkConfig
type SinkConfig struct {
	Protocol                 *string           `json:"protocol,omitempty"`
	SchemaRegistry           *string           `json:"schema_registry,omitempty"`
	CSVConfig                *CSVConfig        `json:"csv,omitempty"`
	DispatchRules            []*DispatchRule   `json:"dispatchers,omitempty"`
	ColumnSelectors          []*ColumnSelector `json:"column_selectors,omitempty"`
	TxnAtomicity             *string           `json:"transaction_atomicity,omitempty"`
	EncoderConcurrency       *int              `json:"encoder_concurrency,omitempty"`
	Terminator               *string           `json:"terminator,omitempty"`
	DateSeparator            *string           `json:"date_separator,omitempty"`
	EnablePartitionSeparator *bool             `json:"enable_partition_separator,omitempty"`
	FileIndexWidth           *int              `json:"file_index_width,omitempty"`
	// deprecated: it's become useless since v9.0.0
	EnableKafkaSinkV2                *bool               `json:"enable_kafka_sink_v2,omitempty"`
	OnlyOutputUpdatedColumns         *bool               `json:"only_output_updated_columns,omitempty"`
	DeleteOnlyOutputHandleKeyColumns *bool               `json:"delete_only_output_handle_key_columns"`
	ContentCompatible                *bool               `json:"content_compatible"`
	OnlyOutputPKColumns              *bool               `json:"only_output_pk_columns,omitempty"`
	SafeMode                         *bool               `json:"safe_mode,omitempty"`
	KafkaConfig                      *KafkaConfig        `json:"kafka_config,omitempty"`
	PulsarConfig                     *PulsarConfig       `json:"pulsar_config,omitempty"`
	MySQLConfig                      *MySQLConfig        `json:"mysql_config,omitempty"`
	CloudStorageConfig               *CloudStorageConfig `json:"cloud_storage_config,omitempty"`
	AdvanceTimeoutInSec              *uint               `json:"advance_timeout,omitempty"`
	SendBootstrapIntervalInSec       *int64              `json:"send_bootstrap_interval_in_sec,omitempty"`
	SendBootstrapInMsgCount          *int32              `json:"send_bootstrap_in_msg_count,omitempty"`
	SendBootstrapToAllPartition      *bool               `json:"send_bootstrap_to_all_partition,omitempty"`
	SendAllBootstrapAtStart          *bool               `json:"send-all-bootstrap-at-start,omitempty"`
	DebeziumDisableSchema            *bool               `json:"debezium_disable_schema,omitempty"`
	DebeziumConfig                   *DebeziumConfig     `json:"debezium,omitempty"`
	OpenProtocolConfig               *OpenProtocolConfig `json:"open,omitempty"`
}

// CSVConfig denotes the csv config
// This is the same as config.CSVConfig
type CSVConfig struct {
	Delimiter            string `json:"delimiter"`
	Quote                string `json:"quote"`
	NullString           string `json:"null"`
	IncludeCommitTs      bool   `json:"include_commit_ts"`
	BinaryEncodingMethod string `json:"binary_encoding_method"`
	OutputOldValue       bool   `json:"output_old_value"`
	OutputHandleKey      bool   `json:"output_handle_key"`
	OutputFieldHeader    bool   `json:"output_field_header"`
}

// LargeMessageHandleConfig denotes the large message handling config
// This is the same as config.LargeMessageHandleConfig
type LargeMessageHandleConfig struct {
	LargeMessageHandleOption      string `json:"large_message_handle_option"`
	LargeMessageHandleCompression string `json:"large_message_handle_compression"`
	ClaimCheckStorageURI          string `json:"claim_check_storage_uri"`
	ClaimCheckRawValue            bool   `json:"claim_check_raw_value"`
}

// DispatchRule represents partition rule for a table
// This is a duplicate of config.DispatchRule
type DispatchRule struct {
	Matcher       []string `json:"matcher,omitempty"`
	PartitionRule string   `json:"partition,omitempty"`
	IndexName     string   `json:"index,omitempty"`
	Columns       []string `json:"columns,omitempty"`
	TopicRule     string   `json:"topic,omitempty"`
}

// ColumnSelector represents a column selector for a table.
// This is a duplicate of config.ColumnSelector
type ColumnSelector struct {
	Matcher []string `json:"matcher,omitempty"`
	Columns []string `json:"columns,omitempty"`
}

// ConsistentConfig represents replication consistency config for a changefeed
// This is a duplicate of config.ConsistentConfig
type ConsistentConfig struct {
	Level                 *string `json:"level,omitempty"`
	MaxLogSize            *int64  `json:"max_log_size,omitempty"`
	FlushIntervalInMs     *int64  `json:"flush_interval,omitempty"`
	MetaFlushIntervalInMs *int64  `json:"meta_flush_interval,omitempty"`
	EncodingWorkerNum     *int    `json:"encoding_worker_num,omitempty"`
	FlushWorkerNum        *int    `json:"flush_worker_num,omitempty"`
	Storage               *string `json:"storage,omitempty"`
	UseFileBackend        *bool   `json:"use_file_backend,omitempty"`
	Compression           *string `json:"compression,omitempty"`
	FlushConcurrency      *int    `json:"flush_concurrency,omitempty"`

	MemoryUsage *ConsistentMemoryUsage `json:"memory_usage,omitempty"`
}

// ConsistentMemoryUsage represents memory usage of Consistent module.
type ConsistentMemoryUsage struct {
	MemoryQuotaPercentage uint64 `json:"memory_quota_percentage"`
}

// ChangefeedSchedulerConfig is per changefeed scheduler settings.
// This is a duplicate of config.ChangefeedSchedulerConfig
type ChangefeedSchedulerConfig struct {
	// EnableTableAcrossNodes set true to split one table to multiple spans and
	// distribute to multiple TiCDC nodes.
	EnableTableAcrossNodes *bool `json:"enable_table_across_nodes,omitempty"`
	// RegionThreshold is the region count threshold of splitting a table.
	RegionThreshold *int `json:"region_threshold,omitempty"`
	// RegionCountPerSpan is the maximax region count for each span when first splitted by RegionCountSpliiter
	RegionCountPerSpan *int `json:"region_count_per_span,omitempty"`
	// RegionCountRefreshInterval controls how often we refresh span region count with PD.
	RegionCountRefreshInterval *time.Duration `json:"region_count_refresh_interval,omitempty"`
	// WriteKeyThreshold is the written keys threshold of splitting a table.
	WriteKeyThreshold *int `json:"write_key_threshold,omitempty"`
	// SchedulingTaskCountPerNode is the upper limit for scheduling tasks each node.
	SchedulingTaskCountPerNode *int `json:"scheduling_task_count_per_node,omitempty"`
	// EnableSplittableCheck controls whether to check if a table is splittable before splitting.
	// If true, only tables with primary key and no unique key can be split.
	// If false, all tables can be split without checking.
	// For MySQL downstream, this is always set to true for data consistency.
	EnableSplittableCheck *bool `json:"enable_splittable_check,omitempty"`
	// ForceSplit controls whether to skip the splittable table check for MySQL downstream.
	// If true, the splittable table check will be skipped even if the downstream is MySQL.
	// This is useful for advanced users who are aware of the risks of splitting unsplittable tables.
	// Default value is false.
	ForceSplit *bool `json:"force_split,omitempty"`
	// These config is used for adjust the frequency of balancing traffic.
	// BalanceScoreThreshold is the score threshold for balancing traffic. Larger value means less frequent balancing.
	BalanceScoreThreshold *int `json:"balance_score_threshold,omitempty"`
	// MinTrafficPercentage is the minimum traffic percentage for balancing traffic. Larger value means less frequent balancing.
	MinTrafficPercentage *float64 `json:"min_traffic_percentage,omitempty"`
	// MaxTrafficPercentage is the maximum traffic percentage for balancing traffic. Less value means less frequent balancing.
	MaxTrafficPercentage *float64 `json:"max_traffic_percentage,omitempty"`
}

// IntegrityConfig is the config for integrity check
// This is a duplicate of Integrity.Config
type IntegrityConfig struct {
	IntegrityCheckLevel   *string `json:"integrity_check_level,omitempty"`
	CorruptionHandleLevel *string `json:"corruption_handle_level,omitempty"`
}

// EtcdData contains key/value pair of etcd data
type EtcdData struct {
	Key   string `json:"key,omitempty"`
	Value string `json:"value,omitempty"`
}

// ResolveLockReq contains request parameter to resolve lock
type ResolveLockReq struct {
	RegionID uint64 `json:"region_id,omitempty"`
	Ts       uint64 `json:"ts,omitempty"`
	PDConfig
}

// ChangeFeedInfo describes the detail of a ChangeFeed
type ChangeFeedInfo struct {
	UpstreamID uint64    `json:"upstream_id,omitempty"`
	ID         string    `json:"id"`
	Keyspace   string    `json:"keyspace"`
	SinkURI    string    `json:"sink_uri,omitempty"`
	CreateTime time.Time `json:"create_time"`
	// Start sync at this commit ts if `StartTs` is specify or using the CreateTime of changefeed.
	StartTs uint64 `json:"start_ts,omitempty"`
	// The ChangeFeed will exits until sync to timestamp TargetTs
	TargetTs uint64 `json:"target_ts,omitempty"`
	// used for admin job notification, trigger watch event in capture
	AdminJobType   config.AdminJobType  `json:"admin_job_type,omitempty"`
	Config         *ReplicaConfig       `json:"config,omitempty"`
	State          config.FeedState     `json:"state,omitempty"`
	Error          *config.RunningError `json:"error,omitempty"`
	CreatorVersion string               `json:"creator_version,omitempty"`

	ResolvedTs     uint64                     `json:"resolved_ts"`
	CheckpointTs   uint64                     `json:"checkpoint_ts"`
	CheckpointTime api.JSONTime               `json:"checkpoint_time"`
	TaskStatus     []config.CaptureTaskStatus `json:"task_status,omitempty"`

	GID            common.GID `json:"gid"`
	MaintainerAddr string     `json:"maintainer_addr,omitempty"`
}

// SyncedStatus describes the detail of a changefeed's synced status
type SyncedStatus struct {
	Synced           bool         `json:"synced"`
	SinkCheckpointTs api.JSONTime `json:"sink_checkpoint_ts"`
	PullerResolvedTs api.JSONTime `json:"puller_resolved_ts"`
	LastSyncedTs     api.JSONTime `json:"last_synced_ts"`
	NowTs            api.JSONTime `json:"now_ts"`
	Info             string       `json:"info"`
}

// toCredential generates a security.Credential from a PDConfig
func (cfg *PDConfig) toCredential() *security.Credential {
	credential := &security.Credential{
		CAPath:   cfg.CAPath,
		CertPath: cfg.CertPath,
		KeyPath:  cfg.KeyPath,
	}
	credential.CertAllowedCN = make([]string, len(cfg.CertAllowedCN))
	copy(credential.CertAllowedCN, cfg.CertAllowedCN)
	return credential
}

// Marshal returns the json marshal format of a ChangeFeedInfo
func (info *ChangeFeedInfo) Marshal() (string, error) {
	data, err := json.Marshal(info)
	return string(data), errors.WrapError(errors.ErrMarshalFailed, err)
}

// Clone returns a cloned ChangeFeedInfo
func (info *ChangeFeedInfo) Clone() (*ChangeFeedInfo, error) {
	s, err := info.Marshal()
	if err != nil {
		return nil, err
	}
	cloned := new(ChangeFeedInfo)
	err = cloned.Unmarshal([]byte(s))
	return cloned, err
}

// Unmarshal unmarshals into *ChangeFeedInfo from json marshal byte slice
func (info *ChangeFeedInfo) Unmarshal(data []byte) error {
	err := json.Unmarshal(data, &info)
	if err != nil {
		return errors.Annotatef(
			errors.WrapError(errors.ErrUnmarshalFailed, err), "Unmarshal data: %v", data)
	}
	return nil
}

// UpstreamConfig contains info to connect to pd
type UpstreamConfig struct {
	ID uint64 `json:"id"`
	PDConfig
}

// ProcessorDetail holds the detail info of a processor
type ProcessorDetail struct {
	// All table ids that this processor are replicating.
	Tables []int64 `json:"table_ids"`
}

// ServerStatus holds some common information of a server
type ServerStatus struct {
	Version   string       `json:"version"`
	GitHash   string       `json:"git_hash"`
	ID        string       `json:"id"`
	ClusterID string       `json:"cluster_id"`
	Pid       int          `json:"pid"`
	IsOwner   bool         `json:"is_owner"`
	Liveness  api.Liveness `json:"liveness"`
}

// Capture holds common information of a capture in cdc
type Capture struct {
	ID string `json:"id"`
	// IsCoordinator is true if the capture is the coordinator of the TiCDC cluster
	// We make its json key as `is-owner` to keep the compatibility with old TiCDC.
	IsCoordinator bool   `json:"is_owner"`
	AdvertiseAddr string `json:"address"`
	ClusterID     string `json:"cluster_id"`
}

// CodecConfig represents a MQ codec configuration
type CodecConfig struct {
	EnableTiDBExtension            *bool   `json:"enable_tidb_extension,omitempty"`
	MaxBatchSize                   *int    `json:"max_batch_size,omitempty"`
	AvroEnableWatermark            *bool   `json:"avro_enable_watermark,omitempty"`
	AvroDecimalHandlingMode        *string `json:"avro_decimal_handling_mode,omitempty"`
	AvroBigintUnsignedHandlingMode *string `json:"avro_bigint_unsigned_handling_mode,omitempty"`
	EncodingFormat                 *string `json:"encoding_format,omitempty"`
}

// PulsarConfig represents a pulsar sink configuration
type PulsarConfig struct {
	TLSKeyFilePath          *string       `json:"tls-certificate-path,omitempty"`
	TLSCertificateFile      *string       `json:"tls-private-key-path,omitempty"`
	TLSTrustCertsFilePath   *string       `json:"tls-trust-certs-file-path,omitempty"`
	PulsarProducerCacheSize *int32        `json:"pulsar-producer-cache-size,omitempty"`
	PulsarVersion           *string       `json:"pulsar-version,omitempty"`
	CompressionType         *string       `json:"compression-type,omitempty"`
	AuthenticationToken     *string       `json:"authentication-token,omitempty"`
	ConnectionTimeout       *int          `json:"connection-timeout,omitempty"`
	OperationTimeout        *int          `json:"operation-timeout,omitempty"`
	BatchingMaxMessages     *uint         `json:"batching-max-messages,omitempty"`
	BatchingMaxPublishDelay *int          `json:"batching-max-publish-delay,omitempty"`
	SendTimeout             *int          `json:"send-timeout,omitempty"`
	TokenFromFile           *string       `json:"token-from-file,omitempty"`
	BasicUserName           *string       `json:"basic-user-name,omitempty"`
	BasicPassword           *string       `json:"basic-password,omitempty"`
	AuthTLSCertificatePath  *string       `json:"auth-tls-certificate-path,omitempty"`
	AuthTLSPrivateKeyPath   *string       `json:"auth-tls-private-key-path,omitempty"`
	OAuth2                  *PulsarOAuth2 `json:"oauth2,omitempty"`
	OutputRawChangeEvent    *bool         `json:"output-raw-change-event,omitempty"`
}

// PulsarOAuth2 is the configuration for OAuth2
type PulsarOAuth2 struct {
	OAuth2IssuerURL  string `json:"oauth2-issuer-url,omitempty"`
	OAuth2Audience   string `json:"oauth2-audience,omitempty"`
	OAuth2PrivateKey string `json:"oauth2-private-key,omitempty"`
	OAuth2ClientID   string `json:"oauth2-client-id,omitempty"`
	OAuth2Scope      string `json:"oauth2-scope,omitempty"`
}

// KafkaConfig represents a kafka sink configuration
type KafkaConfig struct {
	PartitionNum                 *int32                    `json:"partition_num,omitempty"`
	ReplicationFactor            *int16                    `json:"replication_factor,omitempty"`
	KafkaVersion                 *string                   `json:"kafka_version,omitempty"`
	MaxMessageBytes              *int                      `json:"max_message_bytes,omitempty"`
	Compression                  *string                   `json:"compression,omitempty"`
	KafkaClientID                *string                   `json:"kafka_client_id,omitempty"`
	AutoCreateTopic              *bool                     `json:"auto_create_topic,omitempty"`
	DialTimeout                  *string                   `json:"dial_timeout,omitempty"`
	WriteTimeout                 *string                   `json:"write_timeout,omitempty"`
	ReadTimeout                  *string                   `json:"read_timeout,omitempty"`
	RequiredAcks                 *int                      `json:"required_acks,omitempty"`
	SASLUser                     *string                   `json:"sasl_user,omitempty"`
	SASLPassword                 *string                   `json:"sasl_password,omitempty"`
	SASLMechanism                *string                   `json:"sasl_mechanism,omitempty"`
	SASLGssAPIAuthType           *string                   `json:"sasl_gssapi_auth_type,omitempty"`
	SASLGssAPIKeytabPath         *string                   `json:"sasl_gssapi_keytab_path,omitempty"`
	SASLGssAPIKerberosConfigPath *string                   `json:"sasl_gssapi_kerberos_config_path,omitempty"`
	SASLGssAPIServiceName        *string                   `json:"sasl_gssapi_service_name,omitempty"`
	SASLGssAPIUser               *string                   `json:"sasl_gssapi_user,omitempty"`
	SASLGssAPIPassword           *string                   `json:"sasl_gssapi_password,omitempty"`
	SASLGssAPIRealm              *string                   `json:"sasl_gssapi_realm,omitempty"`
	SASLGssAPIDisablePafxfast    *bool                     `json:"sasl_gssapi_disable_pafxfast,omitempty"`
	SASLOAuthClientID            *string                   `json:"sasl_oauth_client_id,omitempty"`
	SASLOAuthClientSecret        *string                   `json:"sasl_oauth_client_secret,omitempty"`
	SASLOAuthTokenURL            *string                   `json:"sasl_oauth_token_url,omitempty"`
	SASLOAuthScopes              []string                  `json:"sasl_oauth_scopes,omitempty"`
	SASLOAuthGrantType           *string                   `json:"sasl_oauth_grant_type,omitempty"`
	SASLOAuthAudience            *string                   `json:"sasl_oauth_audience,omitempty"`
	EnableTLS                    *bool                     `json:"enable_tls,omitempty"`
	CA                           *string                   `json:"ca,omitempty"`
	Cert                         *string                   `json:"cert,omitempty"`
	Key                          *string                   `json:"key,omitempty"`
	InsecureSkipVerify           *bool                     `json:"insecure_skip_verify,omitempty"`
	CodecConfig                  *CodecConfig              `json:"codec_config,omitempty"`
	LargeMessageHandle           *LargeMessageHandleConfig `json:"large_message_handle,omitempty"`
	GlueSchemaRegistryConfig     *GlueSchemaRegistryConfig `json:"glue_schema_registry_config,omitempty"`
	OutputRawChangeEvent         *bool                     `json:"output_raw_change_event,omitempty"`
}

// MySQLConfig represents a MySQL sink configuration
type MySQLConfig struct {
	WorkerCount                  *int    `json:"worker_count,omitempty"`
	MaxTxnRow                    *int    `json:"max_txn_row,omitempty"`
	MaxMultiUpdateRowSize        *int    `json:"max_multi_update_row_size,omitempty"`
	MaxMultiUpdateRowCount       *int    `json:"max_multi_update_row_count,omitempty"`
	TiDBTxnMode                  *string `json:"tidb_txn_mode,omitempty"`
	SSLCa                        *string `json:"ssl_ca,omitempty"`
	SSLCert                      *string `json:"ssl_cert,omitempty"`
	SSLKey                       *string `json:"ssl_key,omitempty"`
	TimeZone                     *string `json:"time_zone,omitempty"`
	WriteTimeout                 *string `json:"write_timeout,omitempty"`
	ReadTimeout                  *string `json:"read_timeout,omitempty"`
	Timeout                      *string `json:"timeout,omitempty"`
	EnableBatchDML               *bool   `json:"enable_batch_dml,omitempty"`
	EnableMultiStatement         *bool   `json:"enable_multi_statement,omitempty"`
	EnableCachePreparedStatement *bool   `json:"enable_cache_prepared_statement,omitempty"`
}

// CloudStorageConfig represents a cloud storage sink configuration
type CloudStorageConfig struct {
	WorkerCount                  *int    `json:"worker_count,omitempty"`
	FlushInterval                *string `json:"flush_interval,omitempty"`
	FileSize                     *int    `json:"file_size,omitempty"`
	OutputColumnID               *bool   `json:"output_column_id,omitempty"`
	FileExpirationDays           *int    `json:"file_expiration_days,omitempty"`
	FileCleanupCronSpec          *string `json:"file_cleanup_cron_spec,omitempty"`
	FlushConcurrency             *int    `json:"flush_concurrency,omitempty"`
	EnableSchemaIndexByGetObject *bool   `json:"enable_schema_index_by_get_object,omitempty"`
	OutputRawChangeEvent         *bool   `json:"output_raw_change_event,omitempty"`
}

// ChangefeedStatus holds common information of a changefeed in cdc
type ChangefeedStatus struct {
	State        string               `json:"state,omitempty"`
	ResolvedTs   uint64               `json:"resolved_ts"`
	CheckpointTs uint64               `json:"checkpoint_ts"`
	LastError    *config.RunningError `json:"last_error,omitempty"`
	LastWarning  *config.RunningError `json:"last_warning,omitempty"`
}

// GlueSchemaRegistryConfig represents a glue schema registry configuration
type GlueSchemaRegistryConfig struct {
	// Name of the schema registry
	RegistryName string `json:"registry_name"`
	// Region of the schema registry
	Region string `json:"region"`
	// AccessKey of the schema registry
	AccessKey string `json:"access_key,omitempty"`
	// SecretAccessKey of the schema registry
	SecretAccessKey string `json:"secret_access_key,omitempty"`
	Token           string `json:"token,omitempty"`
}

// OpenProtocolConfig represents the configurations for open protocol encoding
type OpenProtocolConfig struct {
	OutputOldValue bool `json:"output_old_value"`
}

// DebeziumConfig represents the configurations for debezium protocol encoding
type DebeziumConfig struct {
	OutputOldValue bool `json:"output_old_value"`
}

type DispatcherCount struct {
	Count int `json:"count"`
}

type NodeTableInfo struct {
	NodeID   string  `json:"node_id"`
	TableIDs []int64 `json:"table_ids"`
}

func newNodeTableInfo(nodeID string) *NodeTableInfo {
	return &NodeTableInfo{
		NodeID:   nodeID,
		TableIDs: []int64{},
	}
}

func (t *NodeTableInfo) addTableID(tableID int64) {
	t.TableIDs = append(t.TableIDs, tableID)
}