package service

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

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"go.uber.org/zap"
)

func (s *cdcService) MigrateChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
	rcu uint32,
	displayName string,
) (*vo.ChangefeedMigrationResult, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterID), zap.String("changefeed_id", changefeedID))
	logger.Info("start migrating changefeed to newarch", zap.Uint32("rcu", rcu))

	precheck, bizErr := s.PrecheckMigrateChangefeed(ctx, clusterID, changefeedID, rcu, displayName)
	if bizErr != nil {
		return nil, bizErr
	}
	if !precheck.Passed {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"migration precheck failed: %s",
			failedMigrationChecks(precheck.Checks),
		)
	}

	source, bizErr := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if bizErr != nil {
		return nil, bizErr
	}

	checkpointTSO, bizErr := s.pauseAndCaptureCheckpoint(ctx, source)
	if bizErr != nil {
		return nil, bizErr
	}
	if checkpointTSO == 0 {
		return nil, bizerr.New(bizerr.ErrCodeUnknown,
			"failed to capture checkpoint tso from source changefeed %s", changefeedID)
	}
	logger.Info("captured source checkpoint tso", zap.Uint64("checkpoint_tso", checkpointTSO))

	newVo, bizErr := s.buildMigratedChangefeedVo(ctx, source, rcu, displayName, checkpointTSO)
	if bizErr != nil {
		return nil, bizErr
	}
	created, bizErr := s.CreateChangefeed(ctx, newVo)
	if bizErr != nil {
		logger.Error("failed to create migrated newarch changefeed", zap.Error(bizErr))
		return nil, bizerr.New(
			bizerr.ErrCodeUnknown,
			"source changefeed is paused at checkpoint %d; target creation failed: %s; list changefeeds before resuming the source",
			checkpointTSO,
			bizErr.Error(),
		)
	}
	logger.Info("created newarch changefeed for migration",
		zap.String("new_changefeed_id", created.ChangefeedID),
		zap.Uint64("start_tso", checkpointTSO))
	return &vo.ChangefeedMigrationResult{
		SourceChangefeedID:  source.ID,
		Target:              created,
		SourceCheckpointTSO: checkpointTSO,
	}, nil
}

// PrecheckMigrateChangefeed performs every validation that can safely run
// before the source changefeed is paused. It intentionally does not mutate the
// source state or create a target changefeed.
func (s *cdcService) PrecheckMigrateChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
	rcu uint32,
	displayName string,
) (*vo.ChangefeedMigrationPrecheck, bizerr.BizError) {
	result := &vo.ChangefeedMigrationPrecheck{}
	addCheck := func(name string, err error) bool {
		check := vo.MigrationCheck{Name: name, Passed: err == nil}
		if err != nil {
			check.Message = err.Error()
		}
		result.Checks = append(result.Checks, check)
		return err == nil
	}

	if !addCheck("newarch enabled", func() error {
		if !config.Global().CDCConfig.NewArch.Enabled {
			return errors.New("ticdc newarch is disabled")
		}
		return nil
	}()) {
		return finishMigrationPrecheck(result), nil
	}

	cluster, bizErr := s.getAndCheckCluster(ctx, clusterID)
	if !addCheck("cluster is normal", bizErr) {
		return finishMigrationPrecheck(result), nil
	}
	result.TargetProvider = cluster.GetCloudProvider()
	result.TargetRegion = cluster.GetRegion()
	result.TargetKubernetes = cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	if !addCheck("newarch allowed for cluster", s.validateNewArchAllowedForCluster(cluster)) {
		return finishMigrationPrecheck(result), nil
	}

	source, bizErr := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if !addCheck("source changefeed exists", bizErr) {
		return finishMigrationPrecheck(result), nil
	}
	result.SourceArch = string(source.Arch)
	result.SourceState = string(source.State)
	if !addCheck("source arch is replication_worker", func() error {
		if source.Arch != constant.ChangefeedArchReplicationWorker {
			return fmt.Errorf("actual arch is %s", source.Arch)
		}
		return nil
	}()) {
		return finishMigrationPrecheck(result), nil
	}
	if !addCheck("source control-plane state is running", func() error {
		if source.State != constant.ChangfeedStateRunning {
			return fmt.Errorf("actual state is %s", source.State)
		}
		return nil
	}()) {
		return finishMigrationPrecheck(result), nil
	}

	proxy, bizErr := s.getServerlessClusterProxy(ctx, source.CloudProvider, source.Region, source.Kubernetes)
	if !addCheck("legacy TiCDC is reachable", bizErr) {
		return finishMigrationPrecheck(result), nil
	}
	kernel, bizErr := proxy.GetServerlessChangefeed(ctx, source.KeySpaceID, source.ChangefeedID)
	if bizErr != nil {
		addCheck("source kernel changefeed exists", bizErr)
		return finishMigrationPrecheck(result), nil
	}
	if kernel == nil {
		addCheck("source kernel changefeed exists", errors.New("empty kernel response"))
		return finishMigrationPrecheck(result), nil
	}
	addCheck("source kernel changefeed exists", nil)
	if !addCheck("source kernel state is normal", func() error {
		if kernel.State != serverless.StateNormal {
			return fmt.Errorf("actual kernel state is %s", kernel.State)
		}
		return nil
	}()) {
		return finishMigrationPrecheck(result), nil
	}
	addCheck("source kernel has no error", func() error {
		if kernel.Error != nil && kernel.Error.Message != nil && *kernel.Error.Message != "" {
			return fmt.Errorf("%s", *kernel.Error.Message)
		}
		return nil
	}())
	if kernel.CheckpointTS == nil || *kernel.CheckpointTS == 0 {
		addCheck("source checkpoint is valid", errors.New("checkpoint tso is empty"))
		return finishMigrationPrecheck(result), nil
	}
	result.CurrentCheckpointTSO = *kernel.CheckpointTS
	addCheck("source checkpoint is valid", nil)

	target, buildErr := s.buildMigratedChangefeedVo(ctx, source, rcu, displayName, result.CurrentCheckpointTSO)
	if !addCheck("target configuration can be built", buildErr) {
		return finishMigrationPrecheck(result), nil
	}
	s.normalizeCreateChangefeedDefaults(target, cluster)
	result.TargetRCU = target.RCU
	if !addCheck("target defaults are valid", s.validateCreateChangefeedDefaults(target, cluster)) {
		return finishMigrationPrecheck(result), nil
	}
	_, clientErr := s.getNewArchTiCDCClient(ctx, result.TargetProvider, result.TargetRegion, result.TargetKubernetes)
	if !addCheck("target newarch client can be built", clientErr) {
		return finishMigrationPrecheck(result), nil
	}
	if !addCheck("target configuration is valid", target.Validate(ctx, s.encryptor, cluster)) {
		return finishMigrationPrecheck(result), nil
	}
	if !addCheck("target quota and start position are valid", s.precheckCreate(ctx, target, cluster)) {
		return finishMigrationPrecheck(result), nil
	}
	addCheck("target sink connection succeeds", s.precheckNetworkConnection(ctx, target, cluster))
	return finishMigrationPrecheck(result), nil
}

func finishMigrationPrecheck(result *vo.ChangefeedMigrationPrecheck) *vo.ChangefeedMigrationPrecheck {
	result.Passed = true
	for _, check := range result.Checks {
		if !check.Passed {
			result.Passed = false
			break
		}
	}
	return result
}

func failedMigrationChecks(checks []vo.MigrationCheck) string {
	failed := make([]string, 0)
	for _, check := range checks {
		if !check.Passed {
			failed = append(failed, check.Name+": "+check.Message)
		}
	}
	return strings.Join(failed, "; ")
}

// CompareChangefeeds compares the effective kernel configuration of a paused
// legacy source with a newarch target. Secrets are used for equality checks but
// are never included in the returned messages.
func (s *cdcService) CompareChangefeeds(
	ctx context.Context,
	clusterID, sourceChangefeedID, targetChangefeedID string,
) (*vo.ChangefeedComparison, bizerr.BizError) {
	result := &vo.ChangefeedComparison{}
	addCheck := func(name string, passed bool, message string) {
		result.Checks = append(result.Checks, vo.MigrationCheck{Name: name, Passed: passed, Message: message})
	}

	source, bizErr := s.changefeedRepo.GetById(ctx, clusterID, sourceChangefeedID)
	if bizErr != nil {
		return nil, bizErr
	}
	target, bizErr := s.changefeedRepo.GetById(ctx, clusterID, targetChangefeedID)
	if bizErr != nil {
		return nil, bizErr
	}
	if source.Arch != constant.ChangefeedArchReplicationWorker {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"source changefeed arch is %s, expected replication_worker",
			source.Arch,
		)
	}
	if target.Arch != constant.ChangefeedArchTiCDCNewArch {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"target changefeed arch is %s, expected ticdc_newarch",
			target.Arch,
		)
	}

	legacyProxy, bizErr := s.getServerlessClusterProxy(ctx, source.CloudProvider, source.Region, source.Kubernetes)
	if bizErr != nil {
		return nil, bizErr
	}
	legacyKernel, bizErr := legacyProxy.GetServerlessChangefeed(ctx, source.KeySpaceID, source.ChangefeedID)
	if bizErr != nil {
		return nil, bizErr
	}
	if legacyKernel == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "legacy kernel returned an empty changefeed")
	}

	newArchClient, bizErr := s.getNewArchTiCDCClient(ctx, target.CloudProvider, target.Region, target.Kubernetes)
	if bizErr != nil {
		return nil, bizErr
	}
	shard := s.loadNewArchShardForChangefeed(ctx, target)
	keyspaceName, bizErr := s.getNewArchKeyspaceName(ctx, target.ClusterID)
	if bizErr != nil {
		return nil, bizErr
	}
	newKernel, err := newArchClient.GetChangefeed(ctx, shard.ClusterName, keyspaceName, target.ChangefeedID)
	if err != nil {
		return nil, bizerr.WrapError(err, "get newarch kernel changefeed failed")
	}
	if newKernel == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "newarch kernel returned an empty changefeed")
	}

	result.SourceState = string(legacyKernel.State)
	result.TargetState = string(newKernel.State)
	if legacyKernel.CheckpointTS != nil {
		result.SourceCheckpointTSO = *legacyKernel.CheckpointTS
	}
	result.TargetStartTSO = newKernel.StartTs
	result.TargetCheckpointTSO = newKernel.CheckpointTs

	addCheck("source kernel is paused", legacyKernel.State == serverless.StateStopped,
		fmt.Sprintf("source state is %s", legacyKernel.State))
	addCheck("target kernel is running", string(newKernel.State) == string(serverless.StateNormal),
		fmt.Sprintf("target state is %s", newKernel.State))
	addCheck("target start-ts matches source checkpoint",
		result.SourceCheckpointTSO != 0 && result.TargetStartTSO == result.SourceCheckpointTSO,
		fmt.Sprintf("source checkpoint=%d target start-ts=%d", result.SourceCheckpointTSO, result.TargetStartTSO))
	addCheck("target checkpoint reached start-ts", result.TargetCheckpointTSO >= result.TargetStartTSO,
		fmt.Sprintf("target checkpoint=%d target start-ts=%d", result.TargetCheckpointTSO, result.TargetStartTSO))

	legacySinkURI := ""
	if legacyKernel.SinkURI != nil {
		legacySinkURI = *legacyKernel.SinkURI
	}
	legacyCanonicalURI, legacySafeURI, legacyURIErr := canonicalSinkURI(legacySinkURI)
	newCanonicalURI, newSafeURI, newURIErr := canonicalSinkURI(newKernel.SinkURI)
	uriPassed := legacyURIErr == nil && newURIErr == nil && legacyCanonicalURI == newCanonicalURI
	uriMessage := fmt.Sprintf("source=%s target=%s", legacySafeURI, newSafeURI)
	if legacyURIErr != nil || newURIErr != nil {
		uriMessage = fmt.Sprintf("source parse error=%v target parse error=%v", legacyURIErr, newURIErr)
	}
	addCheck("sink URI matches", uriPassed, uriMessage)

	legacyConfig, err := canonicalKernelConfig(legacyKernel.Config, sinkScheme(legacySinkURI))
	if err != nil {
		return nil, bizerr.WrapError(err, "canonicalize legacy kernel config failed")
	}
	newConfig, err := canonicalKernelConfig(newKernel.Config, sinkScheme(newKernel.SinkURI))
	if err != nil {
		return nil, bizerr.WrapError(err, "canonicalize newarch kernel config failed")
	}
	configMatches := reflect.DeepEqual(legacyConfig, newConfig)
	addCheck("replica configuration matches", configMatches, canonicalConfigDiff(legacyConfig, newConfig))

	result.Passed = true
	for _, check := range result.Checks {
		if !check.Passed {
			result.Passed = false
			break
		}
	}
	return result, nil
}

func canonicalSinkURI(raw string) (canonical, safe string, err error) {
	u, err := url.Parse(raw)
	if err != nil {
		return "", "", err
	}
	if u.Scheme == "" {
		return "", "", errors.New("sink URI scheme is empty")
	}
	query := u.Query()
	for key := range query {
		sort.Strings(query[key])
	}
	u.RawQuery = query.Encode()
	canonical = u.String()
	u.User = nil
	redactedQuery := u.Query()
	for key := range redactedQuery {
		lower := strings.ToLower(key)
		if strings.Contains(lower, "password") || strings.Contains(lower, "secret") ||
			strings.Contains(lower, "token") || strings.Contains(lower, "credential") ||
			strings.Contains(lower, "access-key") || strings.Contains(lower, "access_key") {
			redactedQuery.Set(key, "REDACTED")
		}
	}
	u.RawQuery = redactedQuery.Encode()
	return canonical, u.String(), nil
}

func sinkScheme(raw string) string {
	u, err := url.Parse(raw)
	if err != nil {
		return ""
	}
	return strings.ToLower(u.Scheme)
}

func canonicalKernelConfig(config any, sinkScheme string) (map[string]any, error) {
	data, err := json.Marshal(config)
	if err != nil {
		return nil, err
	}
	result := make(map[string]any)
	if string(data) == "null" {
		return result, nil
	}
	if err := json.Unmarshal(data, &result); err != nil {
		return nil, err
	}
	// These fields are resource/runtime tuning rather than migration semantics
	// and legitimately differ between the dedicated legacy worker and shared
	// newarch TiCDC cluster.
	for _, key := range []string{
		"memory_quota", "mounter", "scheduler", "consistent", "integrity",
		"synced_status", "changefeed_error_stuck_duration", "check_gc_safe_point",
	} {
		delete(result, key)
	}
	for _, key := range []string{"sync_point_interval", "sync_point_retention"} {
		if err := canonicalizeDuration(result, key); err != nil {
			return nil, err
		}
	}
	canonicalizeReplicaDefaults(result)
	canonicalizeSinkConfig(result, sinkScheme)
	return result, nil
}

// canonicalizeReplicaDefaults makes omitted fields equivalent to the defaults
// returned by newer TiCDC APIs. Legacy and newarch changefeeds use different
// API models, so comparing their raw GET responses would treat omitted false
// values and disabled-feature intervals as migration differences.
func canonicalizeReplicaDefaults(config map[string]any) {
	for _, key := range []string{
		"enable_sync_point", "enable_table_monitor", "bdr_mode", "enable_active_active",
	} {
		if _, ok := config[key]; !ok {
			config[key] = false
		}
	}
	if enabled, ok := config["enable_active_active"].(bool); ok && !enabled {
		delete(config, "active_active_progress_interval")
		delete(config, "active_active_sync_stats_interval")
	}
	if enabled, ok := config["enable_sync_point"].(bool); ok && !enabled {
		delete(config, "sync_point_interval")
		delete(config, "sync_point_retention")
	}
}

func canonicalizeDuration(config map[string]any, key string) error {
	value, ok := config[key]
	if !ok {
		return nil
	}
	var duration time.Duration
	switch typed := value.(type) {
	case string:
		parsed, err := time.ParseDuration(typed)
		if err != nil {
			return fmt.Errorf("parse %s duration %q: %w", key, typed, err)
		}
		duration = parsed
	case float64:
		duration = time.Duration(int64(typed))
	default:
		return fmt.Errorf("unsupported %s duration type %T", key, value)
	}
	config[key] = duration.String()
	return nil
}

// canonicalizeSinkConfig removes settings that cannot affect the selected
// sink. TiCDC versions expand different sets of defaults in GET responses. In
// particular, newarch TiCDC returns CSV, encoder, and bootstrap defaults for
// MySQL and cloud-storage changefeeds while legacy TiCDC omits them. Comparing those unrelated
// defaults would reject an otherwise equivalent migration.
func canonicalizeSinkConfig(config map[string]any, scheme string) {
	if scheme != "mysql" && scheme != "tidb" && scheme != "s3" {
		return
	}
	sink, ok := config["sink"].(map[string]any)
	if !ok {
		return
	}
	if scheme == "s3" {
		for _, key := range []string{
			"delete_only_output_handle_key_columns", "content_compatible", "only_output_updated_columns",
		} {
			normalizeBoolDefault(sink, key)
		}
		if csv, ok := sink["csv"].(map[string]any); ok {
			normalizeBoolDefault(csv, "output_field_header")
		}
		if cloudStorage, ok := sink["cloud_storage_config"].(map[string]any); ok {
			normalizeBoolDefault(cloudStorage, "output_column_id")
		}
		delete(sink, "encoder_concurrency")
		return
	}
	allowed := map[string]struct{}{
		"advance_timeout":       {},
		"mysql_config":          {},
		"safe_mode":             {},
		"transaction_atomicity": {},
	}
	for key := range sink {
		if _, ok := allowed[key]; !ok {
			delete(sink, key)
		}
	}
	if _, ok := sink["advance_timeout"]; !ok {
		// TiCDC uses 150 seconds when this field is omitted.
		sink["advance_timeout"] = float64(150)
	}
}

func normalizeBoolDefault(config map[string]any, key string) {
	value, ok := config[key]
	if !ok || value == nil {
		config[key] = false
	}
}

func canonicalConfigDiff(source, target map[string]any) string {
	lines := diffConfigMaps("", source, target)
	if len(lines) == 0 {
		return "canonical configurations match"
	}
	return "canonical configuration diff:\n" + strings.Join(lines, "\n")
}

func diffConfigMaps(path string, source, target map[string]any) []string {
	keys := make(map[string]struct{}, len(source)+len(target))
	for key := range source {
		keys[key] = struct{}{}
	}
	for key := range target {
		keys[key] = struct{}{}
	}
	orderedKeys := make([]string, 0, len(keys))
	for key := range keys {
		orderedKeys = append(orderedKeys, key)
	}
	sort.Strings(orderedKeys)

	lines := make([]string, 0)
	for _, key := range orderedKeys {
		fieldPath := key
		if path != "" {
			fieldPath = path + "." + key
		}
		sourceValue, sourceOK := source[key]
		targetValue, targetOK := target[key]
		if sourceMap, ok := sourceValue.(map[string]any); ok {
			if targetMap, targetMapOK := targetValue.(map[string]any); sourceOK && targetOK && targetMapOK {
				lines = append(lines, diffConfigMaps(fieldPath, sourceMap, targetMap)...)
				continue
			}
		}
		if sourceOK && targetOK && reflect.DeepEqual(sourceValue, targetValue) {
			continue
		}
		lines = append(lines,
			fmt.Sprintf("- %s: %s", fieldPath, formatConfigDiffValue(sourceValue, key, sourceOK)),
			fmt.Sprintf("+ %s: %s", fieldPath, formatConfigDiffValue(targetValue, key, targetOK)),
		)
	}
	return lines
}

func formatConfigDiffValue(value any, key string, present bool) string {
	if !present {
		return "<missing>"
	}
	if isSensitiveConfigKey(key) {
		return "<redacted>"
	}
	data, err := json.Marshal(redactConfigValue(value))
	if err != nil {
		return fmt.Sprintf("<unserializable: %v>", err)
	}
	return string(data)
}

func redactConfigValue(value any) any {
	switch typed := value.(type) {
	case map[string]any:
		redacted := make(map[string]any, len(typed))
		for key, child := range typed {
			if isSensitiveConfigKey(key) {
				redacted[key] = "<redacted>"
				continue
			}
			redacted[key] = redactConfigValue(child)
		}
		return redacted
	case []any:
		redacted := make([]any, len(typed))
		for i, child := range typed {
			redacted[i] = redactConfigValue(child)
		}
		return redacted
	default:
		return value
	}
}

func isSensitiveConfigKey(key string) bool {
	lower := strings.ToLower(key)
	return strings.Contains(lower, "password") || strings.Contains(lower, "secret") ||
		strings.Contains(lower, "token") || strings.Contains(lower, "credential") ||
		strings.Contains(lower, "access-key") || strings.Contains(lower, "access_key")
}

func (s *cdcService) pauseAndCaptureCheckpoint(
	ctx context.Context,
	source *entity.CDCChangefeed,
) (uint64, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", source.ClusterID), zap.String("changefeed_id", source.ID))

	switch source.State {
	case constant.ChangefeedStatePaused:
	case constant.ChangfeedStateRunning, constant.ChangefeedStateWarning:
		if _, err := s.StopChangefeed(ctx, source.ClusterID, source.ID); err != nil {
			return 0, err
		}
		logger.Info("paused source changefeed for migration")
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams,
			"changefeed %s in state %s cannot be migrated", source.ID, source.State)
	}

	proxy, bizErr := s.getServerlessClusterProxy(ctx, source.CloudProvider, source.Region, source.Kubernetes)
	if bizErr != nil {
		return 0, bizErr
	}
	return waitForLegacyChangefeedStopped(ctx, proxy, source, time.Second)
}

func waitForLegacyChangefeedStopped(
	ctx context.Context,
	proxy facade.ServerlessClusterProxyFacade,
	source *entity.CDCChangefeed,
	pollInterval time.Duration,
) (uint64, bizerr.BizError) {
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()

	lastState := "unknown"
	for {
		cf, bizErr := proxy.GetServerlessChangefeed(ctx, source.KeySpaceID, source.ChangefeedID)
		if bizErr != nil {
			return 0, bizErr
		}
		if cf != nil {
			lastState = string(cf.State)
			if cf.State == serverless.StateStopped {
				if cf.CheckpointTS == nil {
					return 0, nil
				}
				return *cf.CheckpointTS, nil
			}
		}

		select {
		case <-ctx.Done():
			return 0, bizerr.New(
				bizerr.ErrCodeUnknown,
				"source changefeed did not reach stopped state before context ended; last state: %s: %v",
				lastState,
				ctx.Err(),
			)
		case <-ticker.C:
		}
	}
}

func (s *cdcService) buildMigratedChangefeedVo(
	ctx context.Context,
	source *entity.CDCChangefeed,
	rcu uint32,
	displayName string,
	startTSO uint64,
) (*vo.Changefeed, bizerr.BizError) {
	if rcu == 0 {
		rcu = source.RCU
	}
	if rcu == 0 {
		rcu = 3
	}
	name := displayName
	if name == "" {
		name = source.Name
	}

	sink, err := s.buildMigratedSinkInfo(ctx, source)
	if err != nil {
		return nil, err
	}
	filter, cloneErr := clonePointer(source.Filter)
	if cloneErr != nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "clone migrated changefeed filter")
	}

	newVo := &vo.Changefeed{
		ClusterID: source.ClusterID,
		Arch:      constant.ChangefeedArchTiCDCNewArch,
		RCU:       rcu,
		Name:      name,
		Sink:      sink,
		Filter:    filter,
		StartPosition: &vo.StartPosition{
			Mode: constant.StartModeFromTSO,
			TSO:  startTSO,
		},
	}
	return newVo, nil
}

func (s *cdcService) buildMigratedSinkInfo(
	ctx context.Context,
	source *entity.CDCChangefeed,
) (*vo.SinkInfo, bizerr.BizError) {
	sink := &vo.SinkInfo{Type: source.SinkType}
	switch source.SinkType {
	case constant.ChangefeedTypeKafka:
		kafka, err := clonePointer(source.Kafka)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "clone migrated kafka sink")
		}
		if kafka != nil && kafka.Authentication.Password != "" {
			password, bizErr := s.decryptMigratedSecret(ctx, "kafka password", kafka.Authentication.Password)
			if bizErr != nil {
				return nil, bizErr
			}
			kafka.Authentication.Password = password
		}
		if kafka != nil && kafka.DataFormat.AvroConfig != nil {
			if registry := kafka.DataFormat.AvroConfig.ConfluentSchemaRegistry; registry != nil &&
				registry.Password != "" {
				password, bizErr := s.decryptMigratedSecret(ctx, "schema registry password", registry.Password)
				if bizErr != nil {
					return nil, bizErr
				}
				registry.Password = password
			}
			if glue := kafka.DataFormat.AvroConfig.AWSGlueSchemaRegistry; glue != nil &&
				glue.SecretAccessKey != "" {
				secret, bizErr := s.decryptMigratedSecret(
					ctx,
					"aws glue schema registry secret access key",
					glue.SecretAccessKey,
				)
				if bizErr != nil {
					return nil, bizErr
				}
				glue.SecretAccessKey = secret
			}
		}
		sink.Kafka = kafka
	case constant.ChangefeedTypeMySQL:
		mysql, err := clonePointer(source.MySQL)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "clone migrated mysql sink")
		}
		if mysql != nil && mysql.Authentication.Password != "" {
			password, bizErr := s.decryptMigratedSecret(ctx, "mysql password", mysql.Authentication.Password)
			if bizErr != nil {
				return nil, bizErr
			}
			mysql.Authentication.Password = password
		}
		sink.MySQL = mysql
	case constant.ChangefeedTypeCloudStorage:
		cloudStorage, err := clonePointer(source.CloudStorageInfo)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "clone migrated cloud storage sink")
		}
		if cloudStorage != nil {
			if s3 := cloudStorage.CloudStorage.S3; s3 != nil && s3.AccessKey != nil &&
				s3.AccessKey.SecretAccessKey != "" {
				secret, bizErr := s.decryptMigratedSecret(ctx, "s3 secret access key", s3.AccessKey.SecretAccessKey)
				if bizErr != nil {
					return nil, bizErr
				}
				s3.AccessKey.SecretAccessKey = secret
			}
			if gcs := cloudStorage.CloudStorage.GCS; gcs != nil && gcs.ServiceAccountKey != "" {
				key, bizErr := s.decryptMigratedSecret(ctx, "gcs service account key", gcs.ServiceAccountKey)
				if bizErr != nil {
					return nil, bizErr
				}
				gcs.ServiceAccountKey = key
			}
			if azure := cloudStorage.CloudStorage.AzureBlob; azure != nil && azure.SasToken != "" {
				token, bizErr := s.decryptMigratedSecret(ctx, "azure blob sas token", azure.SasToken)
				if bizErr != nil {
					return nil, bizErr
				}
				azure.SasToken = token
			}
			if oss := cloudStorage.CloudStorage.OSS; oss != nil && oss.AccessKey != nil &&
				oss.AccessKey.AccessKeySecret != "" {
				secret, bizErr := s.decryptMigratedSecret(ctx, "oss access key secret", oss.AccessKey.AccessKeySecret)
				if bizErr != nil {
					return nil, bizErr
				}
				oss.AccessKey.AccessKeySecret = secret
			}
		}
		sink.CloudStorage = cloudStorage
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported source sink type: %s", source.SinkType)
	}
	return sink, nil
}

func (s *cdcService) decryptMigratedSecret(
	ctx context.Context,
	name string,
	secret string,
) (string, bizerr.BizError) {
	plain, err := s.encryptor.DecryptBase64(secret)
	if err != nil {
		log.WithContextL(ctx).Error("failed to decrypt migrated changefeed secret",
			zap.String("secret_name", name),
			zap.Error(err))
		return "", bizerr.New(bizerr.ErrCodeParams, "failed to decrypt migrated changefeed %s", name)
	}
	return plain, nil
}

func clonePointer[T any](src *T) (*T, error) {
	if src == nil {
		return nil, nil
	}
	data, err := json.Marshal(src)
	if err != nil {
		return nil, err
	}
	var dst T
	if err := json.Unmarshal(data, &dst); err != nil {
		return nil, err
	}
	return &dst, nil
}