package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	filter "github.com/pingcap/tidb/pkg/util/table-filter"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/db"
	"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/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/grpc/grpcli"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	appworkflow "github.com/tidbcloud/serverless-service/internal/workflow"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	DateFormat         = "20060102"
	SyncStateInterval  = 3 * time.Minute
	MaxChangefeedCount = 50

	ChangefeedUnknownError = "unknown error, please create a new one or contact us for help"
)

const (
	QuerySchemaTableReplicationClaus = "SELECT table_schema as 'Database', table_name as 'Table' FROM information_schema.tables WHERE ((table_schema, table_name) IN " +
		"(SELECT DISTINCT table_schema, table_name FROM information_schema.columns WHERE COLUMN_KEY = 'PRI' OR COLUMN_KEY = 'UNI') OR " +
		"(table_schema, table_name) IN (SELECT DISTINCT table_schema, table_name FROM information_schema.tidb_indexes where NON_UNIQUE=0)) AND " +
		"table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT', 'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA', 'METRICS_SCHEMA', 'INSPECTION_SCHEMA', " +
		"'sys', 'mysql', 'heartbeat', 'dm_heartbeat', 'information_schema', 'performance_schema', 'metrics_schema', 'inspection_schema') LIMIT 10000;"
	QuerySchemaTableNonReplicationClaus = "SELECT table_schema as 'Database', table_name as 'Table' FROM information_schema.tables WHERE (table_schema, table_name) NOT IN " +
		"(SELECT DISTINCT table_schema, table_name FROM information_schema.columns WHERE COLUMN_KEY = 'PRI' OR COLUMN_KEY = 'UNI') AND " +
		"(table_schema, table_name) NOT IN (SELECT DISTINCT table_schema, table_name FROM information_schema.tidb_indexes where NON_UNIQUE=0) AND " +
		"table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT', 'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA', 'METRICS_SCHEMA', 'INSPECTION_SCHEMA', " +
		"'sys', 'mysql', 'heartbeat', 'dm_heartbeat', 'information_schema', 'performance_schema', 'metrics_schema', 'inspection_schema') LIMIT 10000;"
)

type CDCService interface {
	ListChangefeeds(ctx context.Context, opts *vo.ListChangefeedsOptions) ([]*vo.Changefeed, uint32, bizerr.BizError)
	GetChangefeed(ctx context.Context, clusterID, changefeedID string) (*vo.Changefeed, bizerr.BizError)
	GetChangefeedByID(ctx context.Context, changefeedID string) (*vo.Changefeed, bizerr.BizError)
	StartChangefeed(ctx context.Context, clusterID, changefeedID string) (*vo.Changefeed, bizerr.BizError)
	StopChangefeed(ctx context.Context, clusterID, changefeedID string) (*vo.Changefeed, bizerr.BizError)
	CreateChangefeed(ctx context.Context, changefeed *vo.Changefeed) (*vo.Changefeed, bizerr.BizError)
	EditChangefeed(ctx context.Context, changefeed *vo.Changefeed) (*vo.Changefeed, bizerr.BizError)
	ApplyChangefeedConfiguration(
		ctx context.Context,
		clusterID, changefeedID, configurationFilename string,
		params map[string]*config_center.ParamItem,
		templateID *int64,
	) bizerr.BizError
	EnsureChangefeedConfiguration(ctx context.Context, clusterID, changefeedID string) bizerr.BizError
	ScaleChangefeed(ctx context.Context, clusterID, changefeedID string, rcu uint32) (*vo.Changefeed, bizerr.BizError)
	MigrateChangefeed(
		ctx context.Context,
		clusterID, changefeedID string,
		rcu uint32,
		displayName string,
	) (*vo.ChangefeedMigrationResult, bizerr.BizError)
	PrecheckMigrateChangefeed(
		ctx context.Context,
		clusterID, changefeedID string,
		rcu uint32,
		displayName string,
	) (*vo.ChangefeedMigrationPrecheck, bizerr.BizError)
	CompareChangefeeds(
		ctx context.Context,
		clusterID, sourceChangefeedID, targetChangefeedID string,
	) (*vo.ChangefeedComparison, bizerr.BizError)
	DeleteChangefeed(ctx context.Context, clusterID, changefeedID string) (*vo.Changefeed, bizerr.BizError)
	TestChangefeed(ctx context.Context, changefeed *vo.Changefeed) bizerr.BizError
	DescribeSchemaTable(
		ctx context.Context,
		clusterId string,
		filterRule []string,
		caseSensitive bool,
	) ([]string, []string, uint64, bizerr.BizError)
	ListChangefeedSpecifications(ctx context.Context, clusterID string) ([]*vo.ChangefeedSpecification, bizerr.BizError)

	// === ops surface ===
	// ListNewArchTiCDCClusters returns ticdc_clusters rows filtered by
	// provider/region/state (empty string means "any"). Used by the
	// ops platform to display the shared TiCDC cluster fleet.
	ListNewArchTiCDCClusters(
		ctx context.Context,
		provider, region, state string,
	) ([]*entity.TiCDCCluster, bizerr.BizError)
	// GetNewArchTiCDCCluster returns one shard row.
	GetNewArchTiCDCCluster(
		ctx context.Context,
		provider, region, kubernetes, shareScope, shareKey string,
	) (*entity.TiCDCCluster, bizerr.BizError)
	// UpgradeNewArchTiCDCCluster writes target_version + state=upgrading
	// to ticdc_clusters and triggers an immediate reconcile so the new
	// version reaches the dataplane on the same call. Returns the
	// updated row.
	UpgradeNewArchTiCDCCluster(
		ctx context.Context,
		provider, region, kubernetes, shareScope, shareKey, targetVersion string,
	) (*entity.TiCDCCluster, bizerr.BizError)
	// ListClusterCDCJobs returns the CDC custom statuses registered for a cluster
	// in tidb-management-service (cluster-level locks and per-changefeed jobs).
	// It is an ops-only helper to inspect residual locks that may block cluster
	// deletion.
	ListClusterCDCJobs(
		ctx context.Context,
		clusterID string,
	) ([]*tidbmgmtservicev1.ClusterCustomStatus, bizerr.BizError)
	// DeleteClusterCDCJob deletes one CDC custom status by its status id. It is the
	// manual ops escape hatch to clear a residual lock that is blocking a user
	// cluster from being deleted.
	DeleteClusterCDCJob(
		ctx context.Context,
		clusterID string,
		statusID uint64,
	) bizerr.BizError
	// GetChangefeedTiCDCClusterNames resolves the shared TiCDC clusters the
	// given ticdc_newarch changefeeds are assigned to, keyed by changefeed
	// id, using bulk per-cluster queries. Recorded assignments are
	// preferred over the config-derived shard. Legacy replication_worker
	// changefeeds are absent from the result. Ops-only display helper.
	GetChangefeedTiCDCClusterNames(
		ctx context.Context,
		clusterID string,
		changefeedIDs []string,
	) (map[string]string, bizerr.BizError)
}

type cdcService struct {
	transProvider               db.TransProvider
	mgmtSvc                     facade.MgmtSvcFacade
	clusterRepo                 repo.ClusterRepo
	changefeedRepo              repo.ChangefeedRepo
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo
	tiCDCClusterRepo            repo.TiCDCClusterRepo
	tiCDCChangefeedRepo         repo.KernelTiCDCChangefeedRepo
	cdcClusterRepo              repo.CDCClusterRepo
	workflow                    appworkflow.Workflow
	clusterProxyProviderFacade  facade.ClusterProxyProviderFacade
	featureGate                 facade.FeatureGate
	accountSvc                  facade.AccountFacade
	encryptor                   facade.EncryptorFacade
	privateLinkConnectionSvc    PrivateLinkConnectionService
	configCenter                grpcli.ConfigCenterClient
	eventbusService             *eventbus.EventbusService
}

func NewCDCService(
	transProvider db.TransProvider,
	mgmtSvc facade.MgmtSvcFacade,
	clusterRepo repo.ClusterRepo,
	changefeedRepo repo.ChangefeedRepo,
	tiCDCChangefeedResourceRepo repo.TiCDCChangefeedResourceRepo,
	tiCDCClusterRepo repo.TiCDCClusterRepo,
	tiCDCChangefeedRepo repo.KernelTiCDCChangefeedRepo,
	cdcClusterRepo repo.CDCClusterRepo,
	workflow appworkflow.Workflow,
	_ *appworkflow.Registrar,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	featureGate facade.FeatureGate,
	accountSvc facade.AccountFacade,
	encryptor facade.EncryptorFacade,
	privateLinkConnectionSvc PrivateLinkConnectionService,
	configCenter grpcli.ConfigCenterClient,
	eventbusService *eventbus.EventbusService,
) CDCService {
	return &cdcService{
		transProvider:               transProvider,
		mgmtSvc:                     mgmtSvc,
		clusterRepo:                 clusterRepo,
		changefeedRepo:              changefeedRepo,
		tiCDCChangefeedResourceRepo: tiCDCChangefeedResourceRepo,
		tiCDCClusterRepo:            tiCDCClusterRepo,
		tiCDCChangefeedRepo:         tiCDCChangefeedRepo,
		cdcClusterRepo:              cdcClusterRepo,
		workflow:                    workflow,
		clusterProxyProviderFacade:  clusterProxyProviderFacade,
		featureGate:                 featureGate,
		accountSvc:                  accountSvc,
		encryptor:                   encryptor,
		privateLinkConnectionSvc:    privateLinkConnectionSvc,
		configCenter:                configCenter,
		eventbusService:             eventbusService,
	}
}

func (s *cdcService) ListChangefeeds(
	ctx context.Context,
	opts *vo.ListChangefeedsOptions,
) ([]*vo.Changefeed, uint32, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", opts.ClusterID))
	logger.Info("list changefeeds", zap.Any("options", opts))

	filter := entity.ListChangefeedsFilter{
		ClusterID: opts.ClusterID,
		Limit:     &opts.Limit,
		Offset:    &opts.Offset,
		OrderBy:   opts.OrderBy,
		SinkType:  opts.SinkType,
	}

	count, err := s.changefeedRepo.CountByFilter(ctx, filter)
	if err != nil {
		logger.Error("count changefeeds failed", zap.Error(err))
		return nil, 0, err
	}

	changefeeds, err := s.changefeedRepo.ListByFilter(ctx, filter)
	if err != nil {
		logger.Error("list changefeeds failed", zap.Error(err))
		return nil, 0, err
	}

	cvs := make([]*vo.Changefeed, 0, len(changefeeds))
	for _, changefeed := range changefeeds {
		cvs = append(cvs, vo.FromEntity(ctx, changefeed))
	}
	return cvs, uint32(count), nil
}

func (s *cdcService) GetChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("changefeedID", changefeedID))
	logger.Info("get changefeed")

	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}

	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) GetChangefeedByID(
	ctx context.Context,
	changefeedID string,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("changefeedID", changefeedID))
	logger.Info("get changefeed by id")

	changefeed, err := s.changefeedRepo.GetByID(ctx, changefeedID)
	if err != nil {
		logger.Error("get changefeed by id failed", zap.Error(err))
		return nil, err
	}

	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) StartChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("changefeedID", changefeedID))
	logger.Info("start changefeed")

	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}
	if err := s.syncChangefeed(ctx, changefeed); err != nil {
		logger.Error("sync changefeeds failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	if changefeed.State != constant.ChangefeedStatePaused && changefeed.State != constant.ChangefeedStateFailed {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"changefeed %s in the state %s is not allowed to resume",
			changefeedID,
			changefeed.State,
		)
	}

	var (
		proxy        facade.ServerlessClusterProxyFacade
		newArchProxy *newArchTiCDCClient
	)
	if isNewArchChangefeed(changefeed) {
		newArchProxy, err = s.getNewArchTiCDCClient(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
	} else {
		proxy, err = s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	}
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return nil, err
	}

	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		changefeed, err = s.changefeedRepo.Start(ctx, clusterID, changefeedID)
		if err != nil {
			logger.Error("start changefeed failed", zap.Error(err))
			if err.ErrCode() == errorcode.AffectedNoRows {
				return bizerr.New(bizerr.ErrCodeParams, "changefeed %s is in the state that not allowed to resume, please check the state", changefeedID)
			}
			return err
		}
		if isNewArchChangefeed(changefeed) {
			clusterName := s.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
			keyspaceName, keyspaceErr := s.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
			if keyspaceErr != nil {
				return keyspaceErr
			}
			if reqErr := newArchProxy.ResumeChangefeed(ctx, clusterName, keyspaceName, changefeed.ChangefeedID, nil); reqErr != nil {
				logger.Error("start newarch cdc changefeed failed", zap.Error(reqErr))
				return bizerr.WrapError(reqErr, "resume newarch changefeed failed")
			}
		} else {
			if err = proxy.ResumeServerlessChangefeed(ctx, changefeed.KeySpaceID, changefeed.ChangefeedID, nil); err != nil {
				logger.Error("start cdc changefeed failed", zap.Error(err))
				return err
			}
		}
		return nil
	}); err != nil {
		return nil, err
	}

	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeStartChangefeed,
		fmt.Sprintf("Changefeed %s started.", changefeed.Name),
	)
	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) StopChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("changefeedID", changefeedID))
	logger.Info("stop changefeed")

	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}
	if err := s.syncChangefeed(ctx, changefeed); err != nil {
		logger.Error("sync changefeeds failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	if !canPauseChangefeed(changefeed.State) {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"changefeed %s can only be paused in running, warning, or running_failed state",
			changefeedID,
		)
	}
	wasFailed := changefeed.State == constant.ChangefeedStateFailed

	var (
		proxy        facade.ServerlessClusterProxyFacade
		newArchProxy *newArchTiCDCClient
	)
	if isNewArchChangefeed(changefeed) {
		newArchProxy, err = s.getNewArchTiCDCClient(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
	} else {
		proxy, err = s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	}
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return nil, err
	}

	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		changefeed, err = s.changefeedRepo.Stop(ctx, clusterID, changefeedID)
		if err != nil {
			logger.Error("stop changefeed failed", zap.Error(err))
			if err.ErrCode() == errorcode.AffectedNoRows {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"changefeed %s can only be paused in running, warning, or running_failed state",
					changefeedID,
				)
			}
			return err
		}
		if isNewArchChangefeed(changefeed) {
			clusterName := s.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
			keyspaceName, keyspaceErr := s.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
			if keyspaceErr != nil {
				return keyspaceErr
			}
			if reqErr := newArchProxy.PauseChangefeed(ctx, clusterName, keyspaceName, changefeed.ChangefeedID); reqErr != nil {
				logger.Error("stop newarch cdc changefeed failed", zap.Error(reqErr))
				return bizerr.WrapError(reqErr, "pause newarch changefeed failed")
			}
		} else if !wasFailed {
			if err = proxy.PauseServerlessChangefeed(ctx, changefeed.KeySpaceID, changefeed.ChangefeedID); err != nil {
				logger.Error("stop cdc changefeed failed", zap.Error(err))
				return err
			}
		} else {
			// Legacy TiCDC ignores AdminStop while a changefeed is failed. Persist the
			// user's desired paused state so they can edit the failed configuration,
			// then recover it through ResumeChangefeed.
			logger.Info("skip no-op legacy pause for failed changefeed")
		}
		return nil
	}); err != nil {
		return nil, err
	}

	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeStopChangefeed,
		fmt.Sprintf("Changefeed %s stopped.", changefeed.Name),
	)
	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) CreateChangefeed(
	ctx context.Context,
	changefeedVo *vo.Changefeed,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeedVo.ClusterID))
	logger.Info("create changefeed",
		zap.String("arch", string(changefeedVo.Arch)),
		zap.Uint32("rcu", changefeedVo.RCU),
		zap.String("sink_type", string(changefeedVo.Sink.Type)))
	// check cluster
	cluster, err := s.getAndCheckCluster(ctx, changefeedVo.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}
	s.normalizeCreateChangefeedDefaults(changefeedVo, cluster)
	if err := s.validateCreateChangefeedDefaults(changefeedVo, cluster); err != nil {
		logger.Error("validate changefeed arch/rcu failed", zap.Error(err))
		return nil, err
	}

	// check changefeed parameters statically
	if err := changefeedVo.Validate(ctx, s.encryptor, cluster); err != nil {
		logger.Error("validate changefeed failed", zap.Error(err))
		return nil, err
	}

	// precheck anything that requires network, such as request db, http, grpc, etc.
	if err := s.precheckCreate(ctx, changefeedVo, cluster); err != nil {
		logger.Error("check if can create changefeed failed", zap.Error(err))
		return nil, err
	}
	if err := s.precheckNetworkConnection(ctx, changefeedVo, cluster); err != nil {
		logger.Error("check network connection failed", zap.Error(err))
		return nil, err
	}

	// build cdc changefeed
	changefeed, err := changefeedVo.ToEntity(ctx, s.encryptor)
	if err != nil {
		logger.Error("convert to CDC entity failed", zap.Error(err))
		return nil, err
	}
	changefeed.State = constant.ChangefeedStateCreating
	changefeed.GenerateName()
	changefeed.FillWithCluster(cluster)

	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		if err = s.tryToRegisterPrivateLinkConnection(ctx, changefeed); err != nil {
			logger.Error("failed to register private link connection", zap.Error(err))
			return err
		}

		changefeed, err = s.changefeedRepo.Create(ctx, changefeed)
		if err != nil {
			logger.Error("create changefeed failed", zap.Error(err))
			return err
		}
		if changefeed.Arch == constant.ChangefeedArchReplicationWorker {
			rowAffected, err := s.cdcClusterRepo.IncreaseOrCreateChangefeed(ctx, changefeed)
			if err != nil {
				logger.Error("failed to increase changefeed", zap.Error(err))
				return err
			}
			if rowAffected == 0 {
				logger.Error("nothing changed when increase changefeed")
				return bizerr.New(bizerr.ErrCodeUnknown, "")
			}
		}
		return nil
	}); err != nil {
		return nil, err
	}
	if err := s.createChangefeedConfigurations(ctx, changefeed); err != nil {
		changefeed.State = constant.ChangefeedStateCreateFailed
		changefeed.Error = err.ErrMsg()
		if _, saveErr := s.changefeedRepo.Save(ctx, changefeed); saveErr != nil {
			logger.Error("mark changefeed config creation failed", zap.Error(saveErr))
		}
		return nil, err
	}
	if isNewArchChangefeed(changefeed) {
		if err := s.submitCreateNewArchChangefeedDag(ctx, changefeed); err != nil {
			changefeed.State = constant.ChangefeedStateCreateFailed
			changefeed.Error = err.ErrMsg()
			if _, saveErr := s.changefeedRepo.Save(ctx, changefeed); saveErr != nil {
				logger.Error("mark newarch changefeed create failed", zap.Error(saveErr))
			}
			return nil, err
		}
	}

	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeCreateChangefeed,
		fmt.Sprintf("Changefeed %s created.", changefeed.Name),
	)
	// convert to vo
	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) tryToRegisterPrivateLinkConnection(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	switch changefeed.SinkType {
	case constant.ChangefeedTypeMySQL:
		network := changefeed.MySQL.Network
		if network.NetworkType == constant.MySQLNetworkTypePrivateLink {
			domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				network.PrivateLinkConnectionId,
				changefeed.ID,
				false,
			)
			if err != nil {
				return err
			}
			changefeed.MySQL.Network.PrivateLinkEndpoint = fmt.Sprintf("%s:%d", domains[0], network.PrivateLinkPort)
		}
	case constant.ChangefeedTypeKafka:
		network := changefeed.Kafka.Network
		if network.NetworkType == constant.KafkaNetworkTypePrivateLink {
			domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				network.PrivateLinkConnectionId,
				changefeed.ID,
				false,
			)
			if err != nil {
				return err
			}
			changefeed.Kafka.Network.PrivateLinkEndpoints = buildPrivateLinkEndpoint(
				domains[0],
				network.PrivateLinkPort,
			)
		}
	}
	return nil
}

func buildPrivateLinkEndpoint(endpoint string, port int32) string {
	if endpoint == "" {
		return ""
	}
	// msk private link connection endpoint already contains port
	if strings.Contains(endpoint, "amazonaws.com") && strings.Contains(endpoint, ":") {
		return endpoint
	}
	return fmt.Sprintf("%s:%d", endpoint, port)
}

// resolvePrivateLinkEndpoints fills in the resolved private-link endpoints
// on the changefeed entity when they are missing — typically because the
// console edit form did not echo them back, and combineChangefeed could not
// safely carry the previously stored values forward (e.g. the user pointed
// the changefeed at a different PrivateLinkConnectionId).
//
// Uses dry-run RegisterPrivateLinkConnection: the (changefeed.ID, plcId) link
// is created exactly once at CreateChangefeed time, and re-creating it would
// fail with ErrCodeRecordExist. Here we only need the resolved domains, not
// a new link row.
func (s *cdcService) resolvePrivateLinkEndpoints(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("clusterID", changefeed.ClusterID)).
		With(zap.String("changefeedID", changefeed.ID))

	switch changefeed.SinkType {
	case constant.ChangefeedTypeKafka:
		if changefeed.Kafka == nil {
			return nil
		}
		network := changefeed.Kafka.Network
		if network.NetworkType != constant.KafkaNetworkTypePrivateLink {
			return nil
		}
		if network.PrivateLinkEndpoints != "" {
			return nil
		}
		// ServerName, when set, replaces the resolved endpoints inside
		// ToChangefeed; resolving here would be wasted work.
		if changefeed.Kafka.Authentication.ServerName != "" {
			return nil
		}
		domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
			ctx,
			changefeed.ClusterID,
			network.PrivateLinkConnectionId,
			"",
			true,
		)
		if err != nil {
			logger.Error("failed to dry-run register kafka private link connection", zap.Error(err))
			return err
		}
		changefeed.Kafka.Network.PrivateLinkEndpoints = buildPrivateLinkEndpoint(
			domains[0], network.PrivateLinkPort,
		)
	case constant.ChangefeedTypeMySQL:
		if changefeed.MySQL == nil {
			return nil
		}
		network := changefeed.MySQL.Network
		if network.NetworkType != constant.MySQLNetworkTypePrivateLink {
			return nil
		}
		if network.PrivateLinkEndpoint != "" {
			return nil
		}
		domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
			ctx,
			changefeed.ClusterID,
			network.PrivateLinkConnectionId,
			"",
			true,
		)
		if err != nil {
			logger.Error("failed to dry-run register mysql private link connection", zap.Error(err))
			return err
		}
		changefeed.MySQL.Network.PrivateLinkEndpoint = buildPrivateLinkEndpoint(
			domains[0], network.PrivateLinkPort,
		)
	}
	return nil
}

func (s *cdcService) tryToDeregisterPrivateLinkConnection(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	switch changefeed.SinkType {
	case constant.ChangefeedTypeMySQL:
		network := changefeed.MySQL.Network
		if network.NetworkType == constant.MySQLNetworkTypePrivateLink {
			err := s.privateLinkConnectionSvc.DeregisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				network.PrivateLinkConnectionId,
				changefeed.ID)
			if err != nil {
				return err
			}
		}
	case constant.ChangefeedTypeKafka:
		network := changefeed.Kafka.Network
		if network.NetworkType == constant.KafkaNetworkTypePrivateLink {
			err := s.privateLinkConnectionSvc.DeregisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				network.PrivateLinkConnectionId,
				changefeed.ID)
			if err != nil {
				return err
			}
		}
	}
	return nil
}

func (s *cdcService) precheckNetworkConnection(
	ctx context.Context,
	changefeedVo *vo.Changefeed,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeedVo.ClusterID))
	logger.Info("pre check changefeed network",
		zap.String("arch", string(changefeedVo.Arch)),
		zap.String("sink_type", string(changefeedVo.Sink.Type)))

	proxy, err := s.getClusterProxyWithTimeout(ctx, cluster)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	// check connect
	switch changefeedVo.Sink.Type {
	case constant.ChangefeedTypeKafka:
		return s.testKafkaConnection(ctx, changefeedVo, proxy)
	case constant.ChangefeedTypeMySQL:
		return s.testMySQLConnection(ctx, changefeedVo, proxy)
	case constant.ChangefeedTypeCloudStorage:
		return s.testCloudStorageConnection(ctx, changefeedVo)
	default:
		return bizerr.New(bizerr.ErrCodeParams, "unsupported sink type: %s", changefeedVo.Sink.Type)
	}
}

func (s *cdcService) precheckCreate(
	ctx context.Context,
	changefeedVo *vo.Changefeed,
	cluster *tidbmgmtservicev1.Cluster,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeedVo.ClusterID))
	logger.Info("pre check for create",
		zap.String("arch", string(changefeedVo.Arch)),
		zap.Uint32("rcu", changefeedVo.RCU),
		zap.String("sink_type", string(changefeedVo.Sink.Type)))

	// check number of changefeeds
	count, err := s.changefeedRepo.CountByFilter(ctx, entity.ListChangefeedsFilter{
		ClusterID: changefeedVo.ClusterID,
	})
	if err != nil {
		logger.Error("failed to count changefeeds", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if count >= MaxChangefeedCount {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"changefeeds number exceeds limit: %d, please contact us for support if you need more quota",
			MaxChangefeedCount)
	}

	serverlessProxy, err := s.getServerlessClusterProxyWithCluster(ctx, cluster)
	if err != nil {
		logger.Error("failed to get serverless cluster proxy", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	// check start position
	sp, err := serverlessProxy.GetServerlessGcSafePoint(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetKeyspaceId(),
	)
	if err != nil {
		logger.Error("failed to get gc safe point", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if err := changefeedVo.StartPosition.Validate(cluster.CreateTime.AsTime(), sp); err != nil {
		return err
	}
	return nil
}

func (s *cdcService) getAndCheckCluster(
	ctx context.Context,
	clusterId string,
) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
	cid, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.InvalidCluster, "invalid clusterID:%s", clusterId)
	}
	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, cid, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		return nil, bizErr
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(errorcode.InvalidCluster, "cluster %s is not in normal status", clusterId)
	}
	return cluster, nil
}

func (s *cdcService) EditChangefeed(
	ctx context.Context,
	changefeedVo *vo.Changefeed,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("clusterID", changefeedVo.ClusterID), zap.String("changefeedID", changefeedVo.ChangefeedID))
	logger.Info("create changefeed", zap.Any("changefeed", changefeedVo))
	// check state
	changefeed, err := s.changefeedRepo.GetById(ctx, changefeedVo.ClusterID, changefeedVo.ChangefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}
	if err := s.syncChangefeed(ctx, changefeed); err != nil {
		logger.Error("sync changefeeds failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if !canEditChangefeed(changefeed.State) {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"changefeed %s can only be edited in paused or running_failed state",
			changefeedVo.ChangefeedID,
		)
	}
	if changefeedVo.Arch != "" && changefeedVo.Arch != changefeed.Arch {
		return nil, bizerr.New(bizerr.ErrCodeParams, "not allowed to change changefeed arch")
	}

	// Reject sink-type / network-topology changes during edit. Switching the
	// network type, private-link connection, or port would also require
	// transactional update of the PrivateLinkConnectionLink rows and a
	// non-dry-run re-resolution of the broker endpoints — explicitly out of
	// scope for the in-place edit path. Customers who need to repoint a
	// changefeed at a different sink should delete and recreate it.
	if err := validateSinkAndNetworkUnchanged(changefeed, changefeedVo); err != nil {
		return nil, err
	}

	cluster, err := s.getAndCheckCluster(ctx, changefeedVo.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return nil, err
	}

	// check changefeed
	if err := changefeedVo.Validate(ctx, s.encryptor, cluster); err != nil {
		logger.Error("validate changefeed failed", zap.Error(err))
		return nil, err
	}

	// precheck network connection
	if err := s.precheckNetworkConnection(ctx, changefeedVo, cluster); err != nil {
		logger.Error("check network connection failed", zap.Error(err))
		return nil, err
	}

	// combine changefeed
	if err := s.combineChangefeed(ctx, changefeed, changefeedVo); err != nil {
		logger.Error("combine changefeed failed", zap.Error(err))
		return nil, err
	}
	if err := entity.ValidateCloudStorageColumnSelectorsArch(changefeed.Arch, changefeed.CloudStorageInfo); err != nil {
		return nil, err
	}

	// If combineChangefeed could not carry the resolved private-link
	// endpoints forward (e.g. the user switched PrivateLinkConnectionId or
	// the stored entity never had them filled), resolve them explicitly
	// before building the SinkURI. Unlike CreateChangefeed we use a dry-run
	// here — the biz id was already linked at create time, so re-registering
	// it would fail with ErrCodeRecordExist.
	if err := s.resolvePrivateLinkEndpoints(ctx, changefeed); err != nil {
		logger.Error("failed to resolve private link endpoints", zap.Error(err))
		return nil, err
	}

	// get cluster proxy url
	var (
		proxy        facade.ServerlessClusterProxyFacade
		newArchProxy *newArchTiCDCClient
	)
	if isNewArchChangefeed(changefeed) {
		newArchProxy, err = s.getNewArchTiCDCClient(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
	} else {
		proxy, err = s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	}
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return nil, err
	}

	changefeedConfig, err := changefeed.ToChangefeedForUpdate(ctx, s.encryptor)
	if err != nil {
		logger.Error("convert to changefeed config failed", zap.Error(err))
		return nil, err
	}
	var (
		legacyEffectiveConfig  *serverless.ChangefeedConfig
		newArchEffectiveConfig *clusterproxy.ChangefeedConfig
		newArchClusterName     string
		newArchKeyspaceName    string
	)
	if isNewArchChangefeed(changefeed) {
		newArchClusterName = s.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
		keyspaceName, keyspaceErr := s.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
		if keyspaceErr != nil {
			return nil, keyspaceErr
		}
		newArchKeyspaceName = keyspaceName
		kernelChangefeed, getErr := newArchProxy.GetChangefeed(
			ctx, newArchClusterName, newArchKeyspaceName, changefeed.ChangefeedID,
		)
		if getErr != nil {
			return nil, bizerr.WrapError(getErr, "get newarch kernel changefeed")
		}
		newArchEffectiveConfig, err = assembleNewArchChangefeedConfig(
			ctx, s.configCenter, changefeed, changefeedConfig, kernelChangefeed.Config,
		)
	} else {
		kernelChangefeed, getErr := proxy.GetServerlessChangefeed(
			ctx, changefeed.KeySpaceID, changefeed.ChangefeedID,
		)
		if getErr != nil {
			return nil, getErr.Chain("get kernel changefeed")
		}
		legacyEffectiveConfig, err = assembleLegacyChangefeedConfig(
			ctx, s.configCenter, changefeed, changefeedConfig, kernelChangefeed.Config,
		)
	}
	if err != nil {
		return nil, err.Chain("assemble effective changefeed config")
	}
	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		if changefeed, err = s.changefeedRepo.Save(ctx, changefeed); err != nil {
			logger.Error("update changefeed failed", zap.Error(err))
			return err
		}

		if isNewArchChangefeed(changefeed) {
			if _, reqErr := newArchProxy.UpdateChangefeed(
				ctx,
				newArchClusterName,
				newArchKeyspaceName,
				changefeed.ChangefeedID,
				newArchEffectiveConfig,
			); reqErr != nil {
				logger.Error("edit newarch cdc changefeed failed", zap.Error(reqErr))
				return bizerr.WrapError(reqErr, "update newarch changefeed failed")
			}
		} else {
			_, err = proxy.UpdateServerlessChangefeed(ctx,
				changefeed.KeySpaceID,
				changefeed.ChangefeedID,
				legacyEffectiveConfig)
			if err != nil {
				logger.Error("edit cdc changefeed failed", zap.Error(err))
				return err
			}
		}
		return nil
	}); err != nil {
		return nil, err
	}
	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeEditChangefeed,
		fmt.Sprintf("Changefeed %s updated.", changefeed.Name),
	)
	// convert to vo
	return vo.FromEntity(ctx, changefeed), nil
}

func canPauseChangefeed(state constant.ChangfeedState) bool {
	return state == constant.ChangfeedStateRunning ||
		state == constant.ChangefeedStateWarning ||
		state == constant.ChangefeedStateFailed
}

func canEditChangefeed(state constant.ChangfeedState) bool {
	return state == constant.ChangefeedStatePaused || state == constant.ChangefeedStateFailed
}

func (s *cdcService) ScaleChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
	rcu uint32,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("changefeedID", changefeedID))
	logger.Info("scale changefeed", zap.Uint32("rcu", rcu))

	if rcu == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "rcu must be greater than 0")
	}

	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}
	if !isNewArchChangefeed(changefeed) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "scale is only supported for ticdc newarch changefeed")
	}
	if err := s.syncChangefeed(ctx, changefeed); err != nil {
		logger.Error("sync changefeeds failed", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if changefeed.State != constant.ChangfeedStateRunning {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"changefeed %s is only allowed to scale in running state",
			changefeedID,
		)
	}
	if changefeed.RCU == rcu {
		return nil, bizerr.New(bizerr.ErrCodeParams, "rcu is not changed")
	}

	changefeed.RCU = rcu
	changefeed.State = constant.ChangefeedStateScaling
	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		if _, saveErr := s.changefeedRepo.Save(ctx, changefeed); saveErr != nil {
			logger.Error("mark newarch changefeed scaling failed", zap.Error(saveErr))
			return saveErr
		}
		if submitErr := s.submitScaleNewArchClusterDag(ctx, changefeed); submitErr != nil {
			logger.Error("submit scale newarch changefeed dag failed", zap.Error(submitErr))
			return submitErr
		}
		return nil
	}); err != nil {
		return nil, err
	}
	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeScaleChangefeed,
		fmt.Sprintf("Changefeed %s scaled to %d RCU.", changefeed.Name, changefeed.RCU),
	)
	return vo.FromEntity(ctx, changefeed), nil
}

// validateSinkAndNetworkUnchanged enforces that EditChangefeed cannot switch
// the sink type or the network-topology identifiers (NetworkType,
// PrivateLinkConnectionId, PrivateLinkPort). These fields are immutable in
// the edit path because changing them would:
//
//   - Leave a stale PrivateLinkConnectionLink row pinning the old PLC
//     (DeleteChangefeed only deregisters the current network.PrivateLinkConnectionId,
//     so the old binding would never be cleaned up — and the old PLC could
//     not be deleted as long as it has dangling links).
//   - Require re-resolving the broker endpoints with the new port and
//     atomically rewriting both the link row and the changefeed entity.
//
// The right user-facing action for repointing a changefeed is delete +
// recreate, which already handles both concerns correctly.
//
// The check tolerates empty values from the request side, because the
// console edit form only echoes back what it knows: a missing field means
// "no change" rather than "set to zero/empty".
func validateSinkAndNetworkUnchanged(
	old *entity.CDCChangefeed,
	edited *vo.Changefeed,
) bizerr.BizError {
	if edited.Sink.Type != old.SinkType {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"not allowed to change sink type from %s to %s during edit",
			old.SinkType, edited.Sink.Type,
		)
	}
	switch old.SinkType {
	case constant.ChangefeedTypeKafka:
		if old.Kafka == nil || edited.Sink.Kafka == nil {
			return nil
		}
		oldN := old.Kafka.Network
		newN := edited.Sink.Kafka.Network
		if newN.NetworkType != "" && newN.NetworkType != oldN.NetworkType {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"not allowed to change kafka network type from %s to %s during edit",
				oldN.NetworkType, newN.NetworkType,
			)
		}
		if oldN.NetworkType == constant.KafkaNetworkTypePrivateLink {
			if newN.PrivateLinkConnectionId != "" &&
				newN.PrivateLinkConnectionId != oldN.PrivateLinkConnectionId {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"not allowed to change kafka private link connection during edit; delete and recreate the changefeed instead",
				)
			}
			if newN.PrivateLinkPort != 0 && newN.PrivateLinkPort != oldN.PrivateLinkPort {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"not allowed to change kafka private link port during edit; delete and recreate the changefeed instead",
				)
			}
		}
	case constant.ChangefeedTypeMySQL:
		if old.MySQL == nil || edited.Sink.MySQL == nil {
			return nil
		}
		oldN := old.MySQL.Network
		newN := edited.Sink.MySQL.Network
		if newN.NetworkType != "" && newN.NetworkType != oldN.NetworkType {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"not allowed to change mysql network type from %s to %s during edit",
				oldN.NetworkType, newN.NetworkType,
			)
		}
		if oldN.NetworkType == constant.MySQLNetworkTypePrivateLink {
			if newN.PrivateLinkConnectionId != "" &&
				newN.PrivateLinkConnectionId != oldN.PrivateLinkConnectionId {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"not allowed to change mysql private link connection during edit; delete and recreate the changefeed instead",
				)
			}
			if newN.PrivateLinkPort != 0 && newN.PrivateLinkPort != oldN.PrivateLinkPort {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"not allowed to change mysql private link port during edit; delete and recreate the changefeed instead",
				)
			}
		}
	case constant.ChangefeedTypeCloudStorage:
		if old.CloudStorageInfo == nil || edited.Sink.CloudStorage == nil {
			return nil
		}
		if edited.Sink.CloudStorage.CloudStorage.Type != old.CloudStorageInfo.CloudStorage.Type {
			return bizerr.New(
				bizerr.ErrCodeParams,
				"not allowed to change cloud storage sink type from %s to %s during edit",
				old.CloudStorageInfo.CloudStorage.Type,
				edited.Sink.CloudStorage.CloudStorage.Type,
			)
		}
	}
	return nil
}

func (s *cdcService) combineChangefeed(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	changefeedVo *vo.Changefeed,
) bizerr.BizError {
	changefeedNew, err := changefeedVo.ToEntity(ctx, s.encryptor)
	if err != nil {
		return err
	}
	// Name is optional, if not provided, use the existing name
	if changefeedNew.Name != "" {
		changefeed.Name = changefeedNew.Name
	}

	// Sink is required, users must provide them completely
	changefeed.SinkType = changefeedNew.SinkType
	switch changefeedNew.SinkType {
	case constant.ChangefeedTypeKafka:
		// The console edit form only echoes the private-link connection id,
		// not the resolved broker endpoints. Carry the previously stored
		// network derivatives forward so that ToChangefeed builds a SinkURI
		// with the real broker hosts instead of "kafka:///topic".
		mergeKafkaNetworkFromOld(changefeedNew.Kafka, changefeed.Kafka)
		changefeed.Kafka = changefeedNew.Kafka
	case constant.ChangefeedTypeMySQL:
		mergeMySQLNetworkFromOld(changefeedNew.MySQL, changefeed.MySQL)
		changefeed.MySQL = changefeedNew.MySQL
	case constant.ChangefeedTypeCloudStorage:
		changefeed.CloudStorageInfo = changefeedNew.CloudStorageInfo
	}
	// Filter is required, users must provide them completely
	changefeed.Filter = changefeedNew.Filter
	return nil
}

// mergeKafkaNetworkFromOld preserves derived private-link/public fields
// (PrivateLinkEndpoints, PrivateLinkPort, PublicEndpoints) on the new
// KafkaInfo from the previously stored one when the new value is empty.
//
// The console edit request only sends the immutable identifiers
// (PrivateLinkConnectionId, NetworkType). The resolved endpoints are
// produced by the backend during changefeed creation and stored in the DB;
// without this merge they are silently wiped, which makes ToChangefeed emit
// an invalid SinkURI like "kafka:///topic" and TiCDC reject the update with
// ErrKafkaNewProducer "missing address".
//
// validateSinkAndNetworkUnchanged already rejects requests that try to
// switch NetworkType / PrivateLinkConnectionId / PrivateLinkPort, so the
// guard branches below are defense-in-depth — if a future caller ever
// reaches this helper without that precheck, it still refuses to splice
// stale endpoints onto a different connection target.
func mergeKafkaNetworkFromOld(newKafka, oldKafka *entity.KafkaInfo) {
	if newKafka == nil || oldKafka == nil {
		return
	}
	// Only carry over if the user did not switch network type or connection.
	if newKafka.Network.NetworkType != oldKafka.Network.NetworkType {
		return
	}
	switch newKafka.Network.NetworkType {
	case constant.KafkaNetworkTypePrivateLink:
		if newKafka.Network.PrivateLinkConnectionId != "" &&
			newKafka.Network.PrivateLinkConnectionId != oldKafka.Network.PrivateLinkConnectionId {
			// User explicitly switched the private link target; do not carry
			// stale endpoints forward — caller will re-resolve.
			return
		}
		if newKafka.Network.PrivateLinkConnectionId == "" {
			newKafka.Network.PrivateLinkConnectionId = oldKafka.Network.PrivateLinkConnectionId
		}
		if newKafka.Network.PrivateLinkEndpoints == "" {
			newKafka.Network.PrivateLinkEndpoints = oldKafka.Network.PrivateLinkEndpoints
		}
		if newKafka.Network.PrivateLinkPort == 0 {
			newKafka.Network.PrivateLinkPort = oldKafka.Network.PrivateLinkPort
		}
	case constant.KafkaNetworkTypePublic:
		if newKafka.Network.PublicEndpoints == "" {
			newKafka.Network.PublicEndpoints = oldKafka.Network.PublicEndpoints
		}
	}
}

// mergeMySQLNetworkFromOld is the MySQL counterpart of
// mergeKafkaNetworkFromOld. See that helper for the rationale; the same
// defense-in-depth comment applies here.
func mergeMySQLNetworkFromOld(newMySQL, oldMySQL *entity.MySQLInfo) {
	if newMySQL == nil || oldMySQL == nil {
		return
	}
	if newMySQL.Network.NetworkType != oldMySQL.Network.NetworkType {
		return
	}
	switch newMySQL.Network.NetworkType {
	case constant.MySQLNetworkTypePrivateLink:
		if newMySQL.Network.PrivateLinkConnectionId != "" &&
			newMySQL.Network.PrivateLinkConnectionId != oldMySQL.Network.PrivateLinkConnectionId {
			return
		}
		if newMySQL.Network.PrivateLinkConnectionId == "" {
			newMySQL.Network.PrivateLinkConnectionId = oldMySQL.Network.PrivateLinkConnectionId
		}
		if newMySQL.Network.PrivateLinkEndpoint == "" {
			newMySQL.Network.PrivateLinkEndpoint = oldMySQL.Network.PrivateLinkEndpoint
		}
		if newMySQL.Network.PrivateLinkPort == 0 {
			newMySQL.Network.PrivateLinkPort = oldMySQL.Network.PrivateLinkPort
		}
	case constant.MySQLNetworkTypePublic:
		if newMySQL.Network.PublicEndpoint == "" {
			newMySQL.Network.PublicEndpoint = oldMySQL.Network.PublicEndpoint
		}
	}
}

func (s *cdcService) DeleteChangefeed(
	ctx context.Context,
	clusterID, changefeedID string,
) (*vo.Changefeed, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", clusterID), zap.String("changefeedID", changefeedID))
	logger.Info("delete changefeed")

	changefeed, err := s.changefeedRepo.GetById(ctx, clusterID, changefeedID)
	if err != nil {
		logger.Error("get changefeed failed", zap.Error(err))
		return nil, err
	}

	if isNewArchChangefeed(changefeed) {
		changefeed.State = constant.ChangefeedStateDeleting
		if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
			if _, saveErr := s.changefeedRepo.Save(ctx, changefeed); saveErr != nil {
				logger.Error("mark newarch changefeed deleting failed", zap.Error(saveErr))
				return saveErr
			}
			if submitErr := s.submitDeleteNewArchChangefeedDag(ctx, changefeed); submitErr != nil {
				logger.Error("submit delete newarch changefeed dag failed", zap.Error(submitErr))
				return submitErr
			}
			return nil
		}); err != nil {
			return nil, err
		}
		s.publishChangefeedEvent(
			ctx,
			changefeed,
			event_constant.EventTypeDeleteChangefeed,
			fmt.Sprintf("Changefeed %s deleted.", changefeed.Name),
		)
		return vo.FromEntity(ctx, changefeed), nil
	}

	proxy, err := s.getServerlessClusterProxy(ctx, changefeed.CloudProvider, changefeed.Region, changefeed.Kubernetes)
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return nil, err
	}

	oldChangefeed := changefeed

	if err := s.transaction(ctx, func(ctx context.Context) bizerr.BizError {
		err := s.tryToDeregisterPrivateLinkConnection(ctx, changefeed)
		if err != nil {
			logger.Error("failed to deregister private link connection", zap.Error(err))
			return err
		}
		changefeed, err = s.changefeedRepo.Delete(ctx, clusterID, changefeedID)
		if err != nil {
			logger.Error("delete changefeed failed", zap.Error(err))
			if err.ErrCode() == errorcode.AffectedNoRows {
				logger.Warn("changefeed already deleted", zap.String("changefeedID", changefeedID))
				changefeed = oldChangefeed
				return nil
			}
			return err
		}

		if err := s.cdcClusterRepo.DecreaseChangefeed(ctx, clusterID); err != nil {
			logger.Error("failed to decrease changefeed", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeUnknown, "")
		}

		if changefeed.State == constant.ChangefeedStateCreating {
			return nil
		}
		if err = proxy.DeleteServerlessChangefeed(ctx, changefeed.KeySpaceID, changefeed.ChangefeedID); err != nil {
			logger.Error("delete cdc changefeed failed", zap.Error(err))
			return err
		}
		return nil
	}); err != nil {
		return nil, err
	}

	s.publishChangefeedEvent(
		ctx,
		changefeed,
		event_constant.EventTypeDeleteChangefeed,
		fmt.Sprintf("Changefeed %s deleted.", changefeed.Name),
	)
	return vo.FromEntity(ctx, changefeed), nil
}

func (s *cdcService) TestChangefeed(ctx context.Context, changefeed *vo.Changefeed) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeed.ClusterID))

	cluster, err := s.getAndCheckCluster(ctx, changefeed.ClusterID)
	if err != nil {
		logger.Error("get cluster failed", zap.Error(err))
		return err
	}

	if err := changefeed.Sink.Validate(ctx, s.encryptor, cluster, true); err != nil {
		logger.Error("validate changefeed sink failed", zap.Error(err))
		return err
	}

	return s.precheckNetworkConnection(ctx, changefeed, cluster)
}

func (s *cdcService) transaction(ctx context.Context, fn func(ctx context.Context) bizerr.BizError) bizerr.BizError {
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		return fn(ctx)
	}); err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return e
		}
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	return nil
}

func (s *cdcService) getServerlessClusterProxy(
	ctx context.Context,
	provider, region, k8s string,
) (facade.ServerlessClusterProxyFacade, bizerr.BizError) {
	clusterProxyUrl := config.GetClusterProxyUrl(provider, region, k8s)
	if clusterProxyUrl == "" {
		regionInfo := fmt.Sprintf("%s.%s.%s", provider, region, k8s)
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster proxy url not found in config for: %s", regionInfo)
	}
	return s.clusterProxyProviderFacade.GetServerlessHTTPProxy(ctx, clusterProxyUrl, nil, nil)
}

func (s *cdcService) getServerlessClusterProxyWithCluster(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (facade.ServerlessClusterProxyFacade, bizerr.BizError) {
	k8s := cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	clusterProxyUrl := config.GetClusterProxyUrl(cluster.GetCloudProvider(), cluster.GetRegion(), k8s)
	if clusterProxyUrl == "" {
		regionInfo := fmt.Sprintf("%s.%s.%s", cluster.GetCloudProvider(), cluster.GetRegion(), k8s)
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster proxy url not found in config for: %s", regionInfo)
	}
	return s.clusterProxyProviderFacade.GetServerlessHTTPProxy(ctx, clusterProxyUrl, nil, nil)
}

// syncChangefeed synchronizes the state of CDC changefeeds with the actual state in the TiCDC clusters.
func (s *cdcService) syncChangefeed(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("clusterID", changefeed.ClusterID), zap.String("changefeedID", changefeed.ID))
	logger.Info("sync changefeed from cdc clusters")

	if changefeed.State == constant.ChangefeedStateCreating ||
		changefeed.State == constant.ChangefeedStateScaling ||
		changefeed.State == constant.ChangefeedStateCreateFailed ||
		changefeed.State == constant.ChangefeedStateDeleting ||
		changefeed.State == constant.ChangefeedStateDeleted {
		logger.Info("changefeed no need to sync, skip")
		return nil
	}

	changefeedMap := make(map[string]*serverless.ChangefeedCommonInfo)
	if isNewArchChangefeed(changefeed) {
		newArchProxy, newArchErr := s.getNewArchTiCDCClient(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
		if newArchErr != nil {
			logger.Error("get newarch ticdc client failed", zap.Error(newArchErr))
			return newArchErr
		}
		clusterName := s.loadNewArchShardForChangefeed(ctx, changefeed).ClusterName
		keyspaceName, keyspaceErr := s.getNewArchKeyspaceName(ctx, changefeed.ClusterID)
		if keyspaceErr != nil {
			logger.Error("get newarch keyspace name failed", zap.Error(keyspaceErr))
			return keyspaceErr
		}
		resp, reqErr := newArchProxy.ListChangefeeds(ctx, clusterName, keyspaceName)
		if reqErr != nil {
			logger.Error("list newarch changefeeds failed", zap.Error(reqErr))
			return bizerr.WrapError(reqErr, "list newarch changefeeds failed")
		}
		logger.Info("list changefeeds from newarch TiCDC", zap.Any("cf", resp.Items))
		for i := range resp.Items {
			item, convErr := convertV2CommonInfoToServerless(&resp.Items[i])
			if convErr != nil {
				return bizerr.WrapError(convErr, "convert newarch changefeed state failed")
			}
			if item != nil && item.ID != nil {
				changefeedMap[*item.ID] = item
			}
		}
	} else {
		proxy, err := s.getServerlessClusterProxy(
			ctx,
			changefeed.CloudProvider,
			changefeed.Region,
			changefeed.Kubernetes,
		)
		if err != nil {
			logger.Error("get serverless cluster proxy failed", zap.Error(err))
			return err
		}
		resp, err := proxy.ListServerlessChangefeeds(ctx, changefeed.KeySpaceID, nil)
		if err != nil {
			logger.Error("list changefeeds failed", zap.Error(err))
			return err
		}
		logger.Info("list changefeeds from TiCDC", zap.Any("cf", resp.Items))
		for _, item := range resp.Items {
			changefeedMap[*item.ID] = item
		}
	}

	SyncFromChangefeedCommon(ctx, changefeed, changefeedMap[changefeedSyncKey(changefeed)])
	if syncErr := s.changefeedRepo.SyncState(ctx, changefeed); syncErr != nil {
		logger.Error(
			"sync changefeed state failed",
			zap.String("clusterID", changefeed.ClusterID),
			zap.String("changefeedID", changefeed.ID),
			zap.Error(syncErr),
		)
		return syncErr
	}

	return nil
}

func (s *cdcService) DescribeSchemaTable(
	ctx context.Context,
	clusterId string,
	filterRule []string,
	caseSensitive bool,
) (replicatedTable, ineligibleTable []string, safepoint uint64, err bizerr.BizError) {
	logger := log.WithContextL(ctx)
	logger.Info("describe schema table", zap.Any("filterRule", filterRule), zap.String("clusterId", clusterId))

	if len(filterRule) == 0 {
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeParams, "filter rule is required")
	}

	// build table f
	f, err1 := filter.Parse(filterRule)
	if err1 != nil {
		logger.Error("parse table filter failed", zap.Error(err1))
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeParams, "invalid filter rule")
	}

	// default is case insensitive
	if !caseSensitive {
		f = filter.CaseInsensitive(f)
	}

	// check cluster
	uint64ClusterId, err1 := strconv.ParseUint(clusterId, 10, 64)
	if err1 != nil {
		logger.Error("failed to parse cluster id", zap.Error(err1))
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	cluster, err := s.mgmtSvc.GetCluster(ctx, uint64ClusterId, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeParams, "cluster is not running")
	}

	// get cluster proxy
	proxy, err := s.getClusterProxyByJWTWithTimeout(ctx, cluster)
	if err != nil {
		logger.Error("failed to get cluster proxy", zap.Error(err))
		return nil, nil, 0, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	// filter table schema, at most 10000 tables will be returned
	replicatedTable, err = queryAndFilterTableSchema(ctx, proxy, f, QuerySchemaTableReplicationClaus)
	if err != nil {
		logger.Error("failed to query replicated schema table", zap.Error(err))
		// Propagate the original error (e.g. a cluster SQL error) so the
		// gateway can surface an actionable message instead of a generic one.
		return nil, nil, 0, err
	}
	ineligibleTable, err = queryAndFilterTableSchema(ctx, proxy, f, QuerySchemaTableNonReplicationClaus)
	if err != nil {
		logger.Error("failed to query ineligible schema table", zap.Error(err))
		return nil, nil, 0, err
	}

	// get serverless cluster proxy
	serverlessProxy, err := s.getServerlessClusterProxyWithCluster(ctx, cluster)
	if err != nil {
		logger.Error("failed to get serverless cluster proxy", zap.Error(err))
		return replicatedTable, ineligibleTable, 0, nil
	}
	sp, err := serverlessProxy.GetServerlessGcSafePoint(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetKeyspaceId(),
	)
	if err != nil {
		logger.Error("failed to get gc safe point", zap.Error(err))
		return replicatedTable, ineligibleTable, 0, nil
	}

	return replicatedTable, ineligibleTable, sp, nil
}

func (s *cdcService) getClusterProxyByJWTWithTimeout(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	timeoutSec := 30
	jwt, err := s.getJwtByCtx(ctx, cluster.ClusterId)
	if err != nil {
		return nil, err
	}
	operator := &clusterproxy.ClusterOperator{
		Username:   jwt.UserName,
		AuthMethod: clusterproxy.AuthMethodToken,
		Credential: jwt.Jwt,
	}
	return s.clusterProxyProviderFacade.GetClusterProxyWithTimeout(ctx, cluster, operator, timeoutSec)
}

func (s *cdcService) getClusterProxyWithTimeout(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	timeoutSec := 30
	return s.clusterProxyProviderFacade.GetClusterProxyWithTimeout(
		ctx,
		cluster,
		&clusterproxy.ClusterOperator{},
		timeoutSec,
	)
}

func (s *cdcService) getJwtByCtx(ctx context.Context, clusterId uint64) (*facade.DBUserJwt, bizerr.BizError) {
	logger := log.WithContextL(ctx)
	switch utils.GetAuthTypeFromContext(ctx) {
	case config.MailAuthType, config.AccessTokenAuthType:
		user, err := utils.GetUserFromContext(ctx)
		if err != nil {
			logger.Error("failed to get user from context", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		jwt, err := s.accountSvc.GetJwtByUser(ctx, user.Email, user.ID, clusterId)
		if err != nil {
			logger.Error("failed to get jwt", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		return jwt, nil
	case config.ApiKeyAuthType:
		apiKey := utils.GetIdentityKeyFromContext(ctx)
		jwt, err := s.accountSvc.GetJwtByApiKey(ctx, apiKey, clusterId)
		if err != nil {
			logger.Error("failed to get jwt", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		return jwt, nil
	default:
		logger.Error("unknown auth type", zap.String("authType", utils.GetAuthTypeFromContext(ctx)))
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported auth method")
	}
}

func queryAndFilterTableSchema(
	ctx context.Context,
	proxy facade.ClusterProxyFacade,
	f filter.Filter,
	sql string,
) ([]string, bizerr.BizError) {
	logger := log.WithContextL(ctx)

	res, err := proxy.QuerySQL(ctx, sql)
	if err != nil {
		logger.Error("failed to query sql", zap.Error(err))
		// Preserve the structured SQL error for safe formatting, but do not mark
		// it user-actionable: this query is owned by the service rather than
		// supplied by the user.
		return nil, err
	}

	result := DatabaseResult{}
	if err1 := json.Unmarshal(res, &result); err1 != nil {
		logger.Error("failed to unmarshal query result", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	// TODO remove it
	logger.Info("query result", zap.Any("result", result))

	filterResult := filterDbTables(&result, f)

	tableSchemas := make([]string, 0, len(filterResult))
	for db, tables := range filterResult {
		for _, table := range tables {
			schema := fmt.Sprintf("%s.%s", db, table)
			tableSchemas = append(tableSchemas, schema)
		}
	}
	return tableSchemas, nil
}

func (s *cdcService) testKafkaConnection(
	ctx context.Context,
	changefeed *vo.Changefeed,
	proxy facade.ClusterProxyFacade,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeed.ClusterID))
	kafkaSink := changefeed.Sink.Kafka

	var endpoints string
	switch kafkaSink.Network.NetworkType {
	case constant.KafkaNetworkTypePublic:
		endpoints = kafkaSink.Network.PublicEndpoints
	case constant.KafkaNetworkTypePrivateLink:
		// use server name as the endpoint if provided.
		// this is only a workaround for server name indication because ticdc does not support specifying server name now.
		// the best way is specify endpoints as url and specify server name separately.
		if kafkaSink.Authentication.ServerName != "" {
			endpoints = fmt.Sprintf("%s:%d", kafkaSink.Authentication.ServerName, kafkaSink.Network.PrivateLinkPort)
			break
		}
		if kafkaSink.Network.PrivateLinkEndpoints == "" {
			domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				kafkaSink.Network.PrivateLinkConnectionId,
				"",
				true,
			)
			if err != nil {
				logger.Error("register private link connection failed", zap.Error(err))
				return err
			}
			endpoints = buildPrivateLinkEndpoint(domains[0], kafkaSink.Network.PrivateLinkPort)
		} else {
			endpoints = kafkaSink.Network.PrivateLinkEndpoints
		}
		// msk private link only supports sasl scram 512 authentication type, without it, test connection will fail
		plc, err := s.privateLinkConnectionSvc.GetPrivateLinkConnection(
			ctx,
			changefeed.ClusterID,
			kafkaSink.Network.PrivateLinkConnectionId,
		)
		if err != nil {
			logger.Error("failed to get private link connection", zap.Error(err))
			return err
		}
		if plc.Type == constant.PrivateLinkConnectionAmazonMSKProvisioned {
			if kafkaSink.Authentication.AuthType != constant.KafkaAuthTypeSASLScram512 {
				return bizerr.New(
					bizerr.ErrCodeParams,
					"MSK private link only supports SCRAM-SHA-512 authentication type",
				)
			}
		}
	}

	connectionModel := clusterproxy.ConnectionModel{
		Type:      string(constant.ChangefeedTypeKafka),
		Url:       endpoints,
		EnableTls: kafkaSink.Authentication.EnableTLS,
	}
	switch kafkaSink.Authentication.AuthType {
	case constant.KafkaAuthTypeDISABLE:
	case constant.KafkaAuthTypeSASLPlain, constant.KafkaAuthTypeSASLScram256, constant.KafkaAuthTypeSASLScram512:
		connectionModel.AuthType = string(toClusterProxyKafkaAuthType(kafkaSink.Authentication.AuthType))
		connectionModel.User = kafkaSink.Authentication.UserName
		connectionModel.Password = kafkaSink.Authentication.Password
	default:
		return bizerr.New(
			bizerr.ErrCodeParams,
			"unsupported kafka authentication type: %s",
			kafkaSink.Authentication.AuthType,
		)
	}

	// Validate runtime params
	connectionModel.KafkaItem = clusterproxy.KafkaItem{
		PartitionNum:      kafkaSink.TopicPartitionConfig.PartitionNum,
		ReplicationFactor: kafkaSink.TopicPartitionConfig.ReplicationFactor,
	}

	logger.Info("start to test kafka connection", zap.Any("connectionModel", connectionModel))
	if err := proxy.TestCDCConnection(ctx, &connectionModel); err != nil {
		logger.Error("test kafka connection failed", zap.Error(err))
		return err
	}
	return nil
}

func (s *cdcService) testMySQLConnection(
	ctx context.Context,
	changefeed *vo.Changefeed,
	proxy facade.ClusterProxyFacade,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeed.ClusterID))
	mysqlSink := changefeed.Sink.MySQL

	var endpoint string
	switch mysqlSink.Network.NetworkType {
	case constant.MySQLNetworkTypePublic:
		endpoint = mysqlSink.Network.PublicEndpoint
	case constant.MySQLNetworkTypePrivateLink:
		if mysqlSink.Network.PrivateLinkEndpoint == "" {
			domains, err := s.privateLinkConnectionSvc.RegisterPrivateLinkConnection(
				ctx,
				changefeed.ClusterID,
				mysqlSink.Network.PrivateLinkConnectionId,
				"",
				true,
			)
			if err != nil {
				logger.Error("register private link connection failed", zap.Error(err))
				return err
			}
			endpoint = fmt.Sprintf("%s:%d", domains[0], mysqlSink.Network.PrivateLinkPort)
		} else {
			endpoint = mysqlSink.Network.PrivateLinkEndpoint
		}
	}

	connectionModel := clusterproxy.ConnectionModel{
		Type:      string(constant.ChangefeedTypeMySQL),
		User:      mysqlSink.Authentication.User,
		Password:  mysqlSink.Authentication.Password,
		Url:       endpoint,
		EnableTls: mysqlSink.Authentication.EnableTLS,
	}

	logger.Info("start to test mysql connection", zap.Any("connectionModel", connectionModel))
	if err := proxy.TestCDCConnection(ctx, &connectionModel); err != nil {
		logger.Error("test mysql connection failed", zap.Error(err))
		return err
	}
	return nil
}

func (s *cdcService) testCloudStorageConnection(ctx context.Context,
	changefeed *vo.Changefeed,
) bizerr.BizError {
	logger := log.WithContextL(ctx).With(zap.String("clusterID", changefeed.ClusterID))
	cloudStorage := changefeed.Sink.CloudStorage.CloudStorage

	regionalCloudStorage := cloudStorage.ToRegionalCloudStorage()
	request := &globalserverlessv1.CheckCloudStoragePermissionRequest{
		CloudStorage: regionalCloudStorage,
		ClusterId:    changefeed.ClusterID,
		FeatureName:  globalserverlessv1.CheckCloudStoragePermissionRequest_FEATURE_NAME_CHANGEFEED,
	}
	logger.Info("start to test cloud storage connection")
	if err := s.mgmtSvc.CheckCloudStoragePermission(ctx, request); err != nil {
		logger.Error("test cloud storage connection failed", zap.Error(err))
		return err
	}
	return nil
}

func toClusterProxyKafkaAuthType(authType constant.KafkaAuthType) constant.ClusterProxyKafkaAuthType {
	switch authType {
	case constant.KafkaAuthTypeSASLPlain:
		return constant.ClusterProxyKafkaAuthTypePlain
	case constant.KafkaAuthTypeSASLScram256:
		return constant.ClusterProxyKafkaAuthTypeScramSha256
	case constant.KafkaAuthTypeSASLScram512:
		return constant.ClusterProxyKafkaAuthTypeScramSha512
	}
	return ""
}

// syncStateFrom The state changes allowed to be synchronized from TiCDC.
var syncStateFromTiCDC = map[constant.ChangfeedState][]string{
	constant.ChangfeedStateRunning: {
		string(constant.ChangefeedStateWarning),
		string(constant.ChangfeedStateRunning),
	},
	constant.ChangefeedStateWarning: {
		string(constant.ChangfeedStateRunning),
		string(constant.ChangefeedStateWarning),
	},
	constant.ChangefeedStateFailed: {
		string(constant.ChangfeedStateRunning),
		string(constant.ChangefeedStateWarning),
		string(constant.ChangefeedStatePaused),
		string(constant.ChangefeedStateFailed),
	},
	constant.ChangefeedStatePaused: {
		string(constant.ChangefeedStatePaused),
	},
}

func SyncFromChangefeedCommon(
	ctx context.Context,
	c *entity.CDCChangefeed,
	changefeed *serverless.ChangefeedCommonInfo,
) {
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", c.ClusterID)).With(zap.String("changefeed_id", c.ID))
	if changefeed == nil {
		c.State = constant.ChangefeedStateDeleted
		logger.Error("changefeed not found in TiCDC")
		metric.ChangefeedStateMisMatchCounter.WithLabelValues(c.ClusterID, c.ChangefeedID, string(c.State), "removed").
			Inc()
		return
	}

	originalState := c.State
	// TiCDC reports a nil or zero checkpoint before the changefeed is initialized.
	// Dereferencing it would panic, and persisting a zero TSO would be read back as
	// the unix epoch, so keep the current value in that case.
	if changefeed.CheckpointTSO != nil && *changefeed.CheckpointTSO > 0 {
		c.CheckpointTSO = *changefeed.CheckpointTSO
	}
	// pending and uninitialized should not happend here.
	// pending will be converted to warning and uninitialized will be converted to normal in TiCDC
	switch changefeed.State {
	case serverless.StateNormal, serverless.StateUnInitialized:
		c.State = constant.ChangfeedStateRunning
		c.Error = ""
	case serverless.StateFailed, serverless.StateWarning, serverless.StatePending:
		logErr := "unknown error"
		if changefeed.Error != nil && changefeed.Error.Message != nil {
			logErr = *changefeed.Error.Message
		}
		logger.Error("changefeed in warning or failed state",
			zap.String("state", string(changefeed.State)),
			zap.String("error", logErr))

		if changefeed.State == serverless.StateFailed && originalState == constant.ChangefeedStatePaused {
			// Legacy TiCDC cannot transition failed -> stopped. Keep the user-requested
			// paused state stable until edit + resume recovers the kernel changefeed.
			c.State = constant.ChangefeedStatePaused
		} else if changefeed.State == serverless.StateFailed {
			c.State = constant.ChangefeedStateFailed
		} else {
			c.State = constant.ChangefeedStateWarning
		}
		errMsg, err := json.Marshal(changefeed.Error)
		if err != nil || len(errMsg) == 0 {
			logger.Error("failed to marshal changefeed error", zap.Error(err))
			c.Error = ChangefeedUnknownError
		} else {
			c.Error = string(errMsg)
		}
	case serverless.StateStopped:
		c.State = constant.ChangefeedStatePaused
		c.Error = ""
		if changefeed.Error != nil && changefeed.Error.Message != nil {
			logger.Error("changefeed paused with error", zap.String("error", *changefeed.Error.Message))
			c.Error = *changefeed.Error.Message
		}
	case serverless.StateRemoved:
		c.State = constant.ChangefeedStateDeleted
		c.Error = ""
	default:
		logger.Error("unsupported changefeed state", zap.String("state", string(changefeed.State)))
		c.State = constant.ChangefeedStateFailed
		c.Error = ChangefeedUnknownError
		metric.ChangefeedStateMisMatchCounter.WithLabelValues(c.ClusterID, c.ChangefeedID, string(originalState), string(changefeed.State)).
			Inc()
		return
	}

	// underlying changefeed is the single source of truth for the state.
	// when unexpected state change happens, we still sync it but will alert it and check it manually.
	allowState := syncStateFromTiCDC[c.State]
	if !slices.Contains(allowState, string(originalState)) {
		logger.Error("illegal state change from TiCDC",
			zap.String("original state", string(originalState)),
			zap.String("new state", string(c.State)))
		metric.ChangefeedStateMisMatchCounter.WithLabelValues(c.ClusterID, c.ChangefeedID, string(originalState), string(changefeed.State)).
			Inc()
	}
}

func (s *cdcService) publishChangefeedEvent(
	ctx context.Context,
	changefeed *entity.CDCChangefeed,
	eventType string,
	summary string,
) {
	if changefeed == nil {
		return
	}
	publishDisplayEventAsync(
		ctx,
		s.eventbusService,
		func(ctx context.Context) *tidbmgmtservicev1.Cluster {
			return loadClusterForEvent(ctx, s.mgmtSvc, changefeed.ClusterID)
		},
		eventType,
		summary,
		event_constant.ChangefeedDetail{
			ID:       changefeed.ID,
			Name:     changefeed.Name,
			SinkType: string(changefeed.SinkType),
			RCU:      changefeed.RCU,
		},
	)
}