package console

import (
	"context"
	"strconv"
	"strings"

	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type ChangefeedController interface {
	dataflow.ConnectorServiceServer
}

type bizChangefeedControllerImpl struct {
	dataflow.UnimplementedConnectorServiceServer
	*BaseController

	changefeedModel   biz_changefeed.BizChangefeedModel
	cdcClusterModel   ticdc.ClusterModel
	downstreamFactory *BizChangefeedDownstreamAdaptorFactory
	featureGate       feature_gate.FeatureGate

	coreCli grpcli.CoreGrpcClient
}

func NewBizChangefeedController(
	trans db.TransProvider, changefeedModel biz_changefeed.BizChangefeedModel,
	cdcClusterModel ticdc.ClusterModel,
	coreCli grpcli.CoreGrpcClient,
	featureGate feature_gate.FeatureGate,
	downstreamFactory *BizChangefeedDownstreamAdaptorFactory,
) ChangefeedController {
	return &bizChangefeedControllerImpl{
		BaseController:    NewBaseController(trans),
		changefeedModel:   changefeedModel,
		cdcClusterModel:   cdcClusterModel,
		downstreamFactory: downstreamFactory,
		coreCli:           coreCli,
		featureGate:       featureGate,
	}
}

func (c *bizChangefeedControllerImpl) ListConnectors(
	ctx context.Context, req *dataflow.ConnectorListReq,
) (*dataflow.Connectors, error) {
	ctx = log.WithTraceID(ctx)
	listReq := &domain.ListBizChangefeedsReq{
		ClusterID: req.ClusterId,
		Page:      req.Page,
		PerPage:   req.PerPage,
	}
	if req.ConnectorType != dataflow.ConnectorType_TOTAL {
		downstreamType := domain.ConnectorType2TiCDCChangefeedDownStreamType(req.ConnectorType)
		listReq.DownstreamType = &downstreamType
	}

	listResp, bizErr := c.changefeedModel.ListBizChangefeeds(ctx, listReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "list biz_changefeeds error")
		return nil, bizErr
	}

	connectors := listResp.ToDataflowConnectors()
	for i := range connectors.Items {
		if bizErr = c.fillCreateReq2Connector(connectors.Items[i], listResp.BizChangefeeds[i]); bizErr != nil {
			logger := log.WithRedactedRequest(log.GetLogger(ctx), req).With(
				zap.Uint64("changefeed_id", listResp.BizChangefeeds[i].BizChangefeedID))
			c.logError(logger, bizErr, "assemble create request error")
			return nil, bizErr
		}
	}

	return &dataflow.Connectors{
		Total: connectors.Total,
		Items: connectors.Items,
	}, nil
}

func (c *bizChangefeedControllerImpl) DeleteConnector(
	ctx context.Context, req *dataflow.ConnectorID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	deleteReq := &domain.DeleteBizChangefeedReq{
		BizChangefeedID: req.ConnectorId,
		UserID:          userID,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	deleteResp, bizErr := c.changefeedModel.DeleteBizChangefeed(ctx, deleteReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "delete biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("delete biz_changefeed successfully", zap.String("workflow_id", deleteResp.WorkflowID))
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) ScaleChangefeed(
	ctx context.Context, req *dataflow.ChangefeedScaleReq,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	scaleReq := &domain.ScaleBizChangefeedReq{
		BizChangefeedID:   req.ChangefeedId,
		SpecificationCode: req.ChangefeedInstanceType,
		UserID:            userID,
	}
	scaleResp, bizErr := c.changefeedModel.ScaleBizChangefeed(ctx, scaleReq)
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	if bizErr != nil {
		c.logError(logger, bizErr, "scale biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("scale biz_changefeed successfully", zap.String("workflow_id", scaleResp.WorkflowID))
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) CalChangefeedFee(
	ctx context.Context, req *dataflow.CalChangefeedFeeReq,
) (*dataflow.CalChangefeedFeeResp, error) {
	ctx = log.WithTraceID(ctx)
	feeReq := &domain.CalBizChangefeedFeeReq{
		TenantID:          req.OrgId,
		ProjectID:         req.ProjectId,
		RegionName:        req.RegionName,
		CloudProvider:     domain.CloudProviderType(req.CloudProviderName),
		SpecificationCode: req.ChangefeedInstanceType,
		Duration:          req.Duration,
	}
	feeResp, bizErr := c.changefeedModel.CalBizChangefeedFee(ctx, feeReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "calculate biz_changefeed fee error")
		return nil, bizErr
	}
	return feeResp.ToDataflowResp(), nil
}

func (c *bizChangefeedControllerImpl) StartConnector(
	ctx context.Context, req *dataflow.ConnectorID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	resumeReq := &domain.ResumeBizChangefeedReq{
		BizChangefeedID: req.ConnectorId,
		UserID:          userID,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	if bizErr := c.changefeedModel.ResumeBizChangefeed(ctx, resumeReq); bizErr != nil {
		c.logError(logger, bizErr, "resume biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("resume biz_changefeed successfully")
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) StopConnector(
	ctx context.Context, req *dataflow.ConnectorID,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	pauseReq := &domain.PauseBizChangefeedReq{
		BizChangefeedID: req.ConnectorId,
		UserID:          userID,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	if bizErr := c.changefeedModel.PauseBizChangefeed(ctx, pauseReq); bizErr != nil {
		c.logError(logger, bizErr, "pause biz_changefeeds error")
		return nil, bizErr
	}

	logger.Info("pause biz_changefeed successfully")
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) GetConnector(
	ctx context.Context, req *dataflow.ConnectorID,
) (*dataflow.Connector, error) {
	ctx = log.WithTraceID(ctx)
	getReq := &domain.GetBizChangefeedReq{
		BizChangefeedID: req.ConnectorId,
	}
	getResp, bizErr := c.changefeedModel.GetBizChangefeed(ctx, getReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get biz_changefeed error")
		return nil, bizErr
	}

	connector := getResp.BizChangefeed.ToDataflowConnector()
	if bizErr = c.fillCreateReq2Connector(connector, getResp.BizChangefeed); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req).With(
			zap.Uint64("changefeed_id", req.GetConnectorId()))
		c.logError(logger, bizErr, "assemble create request error")
		return nil, bizErr
	}

	return connector, nil
}

func (c *bizChangefeedControllerImpl) fillCreateReq2Connector(
	connector *dataflow.Connector, bizChangefeed *domain.BizChangefeed,
) bizerr.BizError {
	downstreamAdaptor, ok := c.downstreamFactory.Get(connector.ConnectorType)
	if !ok {
		return bizerr.New(bizerr.ErrCodeUnknown, "invalid connector type: %d", connector.ConnectorType)
	}
	createReq, bizErr := downstreamAdaptor.AssembleGetResponse(bizChangefeed)
	if bizErr != nil {
		return bizErr
	}
	connector.Request = createReq
	return nil
}

func (c *bizChangefeedControllerImpl) GetClusterTCU(
	ctx context.Context, req *dataflow.GetClusterTCUReq,
) (*dataflow.GetClusterTCUResp, error) {
	ctx = log.WithTraceID(ctx)
	rcuReq := &domain.GetClusterBizChangefeedRCUReq{
		ClusterID: req.ClusterId,
	}
	rcuResp, bizErr := c.changefeedModel.GetClusterBizChangefeedRCU(ctx, rcuReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "get cluster biz_changefeed rcu error")
		return nil, bizErr
	}
	return rcuResp.ToDataflowResp(), nil
}

func (c *bizChangefeedControllerImpl) CreateConnector(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (*dataflow.ConnectorID, error) {
	ctx = log.WithTraceID(ctx)
	if err := c.mustNotReplicate2Itself(ctx, req.ClusterId, req.ConnectorType, req.MysqlInfo); err != nil {
		return nil, err
	}
	logger := log.GetLogger(ctx)
	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ConnectorType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid connector type: %d", req.ConnectorType)
	}

	cluster, bizErr := c.coreCli.GetAvailableClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		c.logError(log.WithRedactedRequest(logger, req), bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	deployType := getBizChangefeedDeployType(c.featureGate, cluster, req.ConnectorType)
	downstreamParams, bizErr := downstreamAdaptor.ParseCreateParams(ctx, req)
	if bizErr != nil {
		c.logError(log.WithRedactedRequest(logger, req), bizErr, "parse param error")
		return nil, bizErr
	}
	ticdcVersion, newArch, bizErr := c.getTiCDCClusterVersionForNewChangefeed(
		ctx, &dataflow.ListConnectorEnabledFeaturesReq{
			OrgId:         req.OrgId,
			ProjectId:     req.ProjectId,
			ClusterId:     req.ClusterId,
			ConnectorType: &req.ConnectorType,
		})
	if bizErr != nil {
		c.logError(log.WithRedactedRequest(logger, req), bizErr, "get ticdc cluster version error")
		return nil, bizErr
	}
	if bizErr := downstreamAdaptor.ValidateCreateParams(
		downstreamParams, cluster, ticdcVersion, newArch); bizErr != nil {
		c.logError(log.WithRedactedRequest(logger, req), bizErr, "validate downstream params error")
		return nil, bizErr
	}

	createReq := c.assembleCreateBizChangefeedReq(ctx, req, downstreamParams, deployType)
	createResp, bizErr := c.changefeedModel.CreateBizChangefeed(ctx, createReq)
	if bizErr != nil {
		c.logError(log.WithRedactedRequest(logger, req), bizErr, "create biz_changefeed error")
		return nil, bizErr
	}

	return &dataflow.ConnectorID{
		ConnectorId: createResp.BizChangefeedID,
		ClusterId:   req.ClusterId,
		ProjectId:   req.ProjectId,
		OrgId:       req.OrgId,
	}, nil
}

func (c *bizChangefeedControllerImpl) assembleCreateBizChangefeedReq(
	ctx context.Context,
	req *dataflow.ConnectorCreateReq,
	downstreamParams interface{},
	deployType entity.TiCDCClusterDeployType,
) *domain.CreateBizChangefeedReq {
	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	createReq := &domain.CreateBizChangefeedReq{
		Name:                req.Name,
		ClusterID:           req.ClusterId,
		DownstreamType:      req.ConnectorType,
		NetworkInfo:         req.NetworkInfo,
		DownstreamParameter: downstreamParams,
		SpecificationCode:   req.ChangefeedInstanceType,
		UserID:              userID,
		DeployType:          deployType,
		DryRun:              req.DryRun,
	}
	if req.NetworkInfo != nil && req.NetworkInfo.SinkEndpointId != nil {
		createReq.PrivateLinkEndpointID = entity.ID(*req.NetworkInfo.SinkEndpointId)
	}
	return createReq
}

func (c *bizChangefeedControllerImpl) EditConnector(
	ctx context.Context, req *dataflow.ConnectorEditReq,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if err := c.mustNotReplicate2Itself(ctx, req.ClusterId, req.ConnectorType, req.MysqlInfo); err != nil {
		return nil, err
	}
	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ConnectorType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid connector type: %d", req.ConnectorType)
	}

	ticdcVersion, newArch, bizErr := c.getTiCDCVersionByChangefeed(ctx, req.ConnectorId)
	if bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "get ticdc cluster version error")
		return nil, bizErr
	}
	downstreamParams := downstreamAdaptor.ParseEditParams(req)
	if bizErr := downstreamAdaptor.ValidateEditParams(downstreamParams, ticdcVersion, newArch); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "validate downstream params error")
		return nil, bizErr
	}

	userID := bizutils.GetUIDOrZeroFromContext(ctx)
	editReq := &domain.EditBizChangefeedReq{
		BizChangefeedID:     req.ConnectorId,
		Name:                req.Name,
		DownstreamType:      req.ConnectorType,
		DownstreamParameter: downstreamParams,
		UserID:              userID,
		DryRun:              req.DryRun,
	}
	if bizErr := c.changefeedModel.EditBizChangefeed(ctx, editReq); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "edit biz_changefeed error")
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) TestConnector(
	ctx context.Context, req *dataflow.ConnectorCreateReq,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	if err := c.mustNotReplicate2Itself(ctx, req.ClusterId, req.ConnectorType, req.MysqlInfo); err != nil {
		return nil, err
	}
	downstreamAdaptor, ok := c.downstreamFactory.Get(req.ConnectorType)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid connector type: %d", req.ConnectorType)
	}

	downstreamParams, bizErr := downstreamAdaptor.ParseCreateParams(ctx, req)
	if bizErr != nil {
		return nil, bizErr
	}
	testReq := &domain.TestBizChangefeedReq{
		ClusterID:           req.ClusterId,
		DownstreamType:      req.ConnectorType,
		DownstreamParameter: downstreamParams,
	}
	if bizErr := c.changefeedModel.TestBizChangefeed(ctx, testReq); bizErr != nil {
		logger := log.WithRedactedRequest(log.GetLogger(ctx), req)
		c.logError(logger, bizErr, "test biz_changefeed error")
		return nil, bizErr
	}
	return &emptypb.Empty{}, nil
}

func (c *bizChangefeedControllerImpl) GetClusterTCUForAdmin(
	ctx context.Context, req *dataflow.GetClusterTCUReq,
) (*dataflow.GetClusterTCUResp, error) {
	// admin permission is check in func canAccessAdminEndpoint
	return c.GetClusterTCU(ctx, req)
}

func (c *bizChangefeedControllerImpl) GetChangefeedDetail(ctx context.Context,
	req *dataflow.ConnectorID,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	detailReq := &domain.GetBizChangefeedDetailReq{
		BizChangefeedID: req.ConnectorId,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.GetChangefeedDetail(ctx, detailReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get biz_changefeeds configuration detail error")
		return nil, bizErr
	}

	logger.Info("get biz_changefeeds configuration detail successfully")
	return resp.Changefeed, nil
}

func (c *bizChangefeedControllerImpl) GetTiCDCCluster(ctx context.Context,
	req *dataflow.ConnectorID,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	clusterReq := &domain.GetBizChangefeedClusterReq{
		BizChangefeedID: req.ConnectorId,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.GetTiCDCCluster(ctx, clusterReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get biz_changefeeds cluster detail error")
		return nil, bizErr
	}

	logger.Info("get biz_changefeeds cluster detail successfully")
	return resp.TiCDCCluster, nil
}

func (c *bizChangefeedControllerImpl) ListTiCDCClusters(ctx context.Context,
	req *dataflow.ClusterIdReq,
) (*structpb.Struct, error) {
	ctx = log.WithTraceID(ctx)
	cluster, bizErr := c.coreCli.GetAvailableClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
		c.logError(logger, bizErr, "get cluster error")
		return nil, bizErr.Chain("get cluster")
	}
	clusterReq := &domain.ListTiCDCClusterReq{
		Cluster: cluster,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	resp, bizErr := c.changefeedModel.ListTiCDCClusters(ctx, clusterReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "get ticdc clusters failed")
		return nil, bizErr
	}
	logger.Info("get ticdc clusters detail successfully")
	return resp.TiCDCClusters, nil
}

func (c *bizChangefeedControllerImpl) DescribeAdminProfile(
	ctx context.Context, req *dataflow.ClusterIdReq,
) (*dataflow.AdminProfileResp, error) {
	ctx = log.WithTraceID(ctx)
	profileReq := &domain.DescribeAdminProfileReq{
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
		ClusterID: req.ClusterId,
	}
	profileResp, bizErr := c.changefeedModel.DescribeAdminProfile(ctx, profileReq)
	if bizErr != nil {
		logger := log.GetLogger(ctx).With(zap.Any("req", req))
		c.logError(logger, bizErr, "describe admin profile error")
		return nil, bizErr
	}
	return &dataflow.AdminProfileResp{
		Type:          string(profileResp.InfraProviderType),
		AwsAccountId:  profileResp.AWSAccountID,
		AwsExternalId: profileResp.AWSExternalID,
		GcpAccountId:  profileResp.GCPServiceAccount,
	}, nil
}

func (c *bizChangefeedControllerImpl) ListConnectorEnabledFeatures(
	ctx context.Context, req *dataflow.ListConnectorEnabledFeaturesReq,
) (*dataflow.ListConnectorEnabledFeaturesResp, error) {
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	var (
		newArch             bool
		ticdcClusterVersion string
		bizErr              bizerr.BizError
	)
	if req.ConnectorId != nil {
		// changefeed existing, for editing changefeed
		ticdcClusterVersion, newArch, bizErr = c.getTiCDCVersionByChangefeed(ctx, *req.ConnectorId)
		if bizErr != nil {
			c.logError(logger, bizErr, "get cdc version by changefeed error")
			return nil, bizErr
		}
	} else {
		ticdcClusterVersion, newArch, bizErr = c.getTiCDCClusterVersionForNewChangefeed(ctx, req)
		if bizErr != nil {
			c.logError(logger, bizErr, "get cdc version for new changefeed error")
			return nil, bizErr
		}
	}

	features, err := internal_config.Global().ChangefeedConfig.GetEnabledFeatures(
		utils.NormalizeVersion(ticdcClusterVersion), newArch)
	if err != nil {
		bizErr := bizerr.WrapErrorWithCode(err, bizerr.ErrCodeUnknown, "get changefeed enabled features error")
		c.logError(logger, bizErr, "get changefeed enabled features error")
		return nil, bizErr
	}
	return &dataflow.ListConnectorEnabledFeaturesResp{
		Features: features,
	}, nil
}

func (c *bizChangefeedControllerImpl) getTiCDCClusterVersionForNewChangefeed(
	ctx context.Context, req *dataflow.ListConnectorEnabledFeaturesReq,
) (version string, newArch bool, err bizerr.BizError) {
	cluster, bizErr := c.coreCli.GetAvailableClusterByID(ctx, req.ClusterId)
	if bizErr != nil {
		return "", false, bizErr.Chain("get cluster error")
	}

	deployType := entity.TiCDCClusterDeployTypeShared
	if req.ConnectorType != nil {
		deployType = getBizChangefeedDeployType(c.featureGate, cluster, *req.ConnectorType)
	}

	if deployType == entity.TiCDCClusterDeployTypeShared {
		// shared ticdc cluster may be existing
		version, newArch, bizErr = c.getExistingSharedTiCDCClusterVersion(ctx, req.ClusterId)
		if bizErr != nil {
			if bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				return "", false, bizErr.Chain("get existing shared ticdc cluster version error")
			}
		} else {
			return version, newArch, nil
		}
	}

	// no existing ticdc cluster, get new ticdc cluster version
	getReq := &domain.GetNewTiCDCClusterVersionReq{
		TenantID:           req.OrgId,
		ProjectID:          req.ProjectId,
		TiDBClusterID:      req.ClusterId,
		TiDBClusterVersion: cluster.Version,
	}
	getResp, bizErr := c.cdcClusterModel.GetNewClusterVersion(ctx, getReq)
	if bizErr != nil {
		return "", false, bizErr.Chain("get new ticdc cluster version error")
	}
	newArch = getResp.NewArch
	version = getResp.Version
	return version, newArch, nil
}

func (c *bizChangefeedControllerImpl) getTiCDCVersionByChangefeed(
	ctx context.Context, changefeedID uint64,
) (version string, newArch bool, err bizerr.BizError) {
	changefeedReq := &domain.GetBizChangefeedReq{
		BizChangefeedID: changefeedID,
	}
	changefeedResp, bizErr := c.changefeedModel.GetBizChangefeed(ctx, changefeedReq)
	if bizErr != nil {
		return "", false, bizErr.Chain("get biz_changefeed error")
	}
	ticdcClusterID := changefeedResp.BizChangefeed.TiCDCClusterID

	cdcReq := &domain.GetExistingTiCDCClusterVersionReq{
		TiCDCClusterID: ticdcClusterID,
	}
	cdcResp, bizErr := c.cdcClusterModel.GetExistingClusterVersion(ctx, cdcReq)
	if bizErr != nil {
		return "", false, bizErr.Chain("get existing ticdc cluster version error")
	}
	return cdcResp.Version, cdcResp.NewArch, nil
}

func (c *bizChangefeedControllerImpl) getExistingSharedTiCDCClusterVersion(
	ctx context.Context, tidbClusterID uint64,
) (version string, newArch bool, err bizerr.BizError) {
	listReq := &domain.ListTiCDCClustersReq{
		TiDBClusterID: tidbClusterID,
	}
	listResp, bizErr := c.cdcClusterModel.List(ctx, listReq)
	if bizErr != nil {
		return "", false, bizErr.Chain("list ticdc clusters error")
	}

	var ticdcClusterID uint64
	for _, ticdcCluster := range listResp.Items {
		if ticdcCluster.Status == entity.TiCDCClusterStatusDeleting ||
			ticdcCluster.Status == entity.TiCDCClusterStatusDeleted {
			continue
		}
		if ticdcCluster.DeployType == entity.TiCDCClusterDeployTypeShared {
			ticdcClusterID = ticdcCluster.ID.Uint64()
			break
		}
	}
	if ticdcClusterID == 0 {
		return "", false, bizerr.New(bizerr.ErrCodeRecordNotFound, "no shared ticdc cluster found")
	}

	cdcReq := &domain.GetExistingTiCDCClusterVersionReq{
		TiCDCClusterID: ticdcClusterID,
	}
	cdcResp, bizErr := c.cdcClusterModel.GetExistingClusterVersion(ctx, cdcReq)
	if bizErr != nil {
		return "", false, bizErr.Chain("get existing ticdc cluster version error")
	}
	return cdcResp.Version, cdcResp.NewArch, nil
}

func (c *bizChangefeedControllerImpl) DescribeSchemaTable(
	ctx context.Context, req *dataflow.DescribeSchemaTableReq,
) (*dataflow.DescribeSchemaTableResp, error) {
	ctx = log.WithTraceID(ctx)
	describeReq := &domain.DescribeSchemaTableReq{
		ClusterID:     req.ClusterId,
		Filters:       req.FilterRule,
		CaseSensitive: req.CaseSensitive,
	}
	logger := log.GetLogger(ctx).With(zap.Any("req", req))
	mresp, bizErr := c.changefeedModel.DescribeSchemaTable(ctx, describeReq)
	if bizErr != nil {
		c.logError(logger, bizErr, "describe schema error")
		return nil, bizErr
	}
	logger.Info("describe schema table successfully")
	return &dataflow.DescribeSchemaTableResp{
		ClusterId:                 req.ClusterId,
		ProjectId:                 req.ProjectId,
		ReplicateDenySchemaTable:  mresp.ReplicateDenySchemaTable,
		ReplicateAllowSchemaTable: mresp.ReplicateAllowSchemaTable,
		GcSafePoint:               timestamppb.New(utils.GetTimeFromTS(mresp.GcSafePoint)),
	}, nil
}

func (c *bizChangefeedControllerImpl) mustNotReplicate2Itself(
	ctx context.Context, clusterID uint64, changefeedType dataflow.ConnectorType, mysqlInfo *dataflow.MySQLInfo,
) error {
	if changefeedType != dataflow.ConnectorType_MYSQL {
		return nil
	}
	if mysqlInfo == nil || mysqlInfo.Connection == nil || mysqlInfo.Connection.Endpoint == "" {
		// maybe private link or invalid input, other validation logic will validate it.
		return nil
	}

	mysqlHost, mysqlPort, err := parseMySQLEndpoint(mysqlInfo.Connection.Endpoint)
	if err != nil {
		return err
	}

	accessPoints, bizErr := c.coreCli.GetClusterAccessPoints(ctx, clusterID)
	if bizErr != nil {
		log.GetLogger(ctx).Error(
			"GetClusterAccessPoints error", zap.Uint64("cluster_id", clusterID), zap.Error(bizErr))
		return bizerr_wrapper.NewInternalErr("Unexpected internal error, please retry later. " +
			"If problem can not be solved, please contact technical support.")
	}

	if isOneOfAccessPoints(accessPoints, mysqlHost, mysqlPort) {
		return bizerr_wrapper.NewInvalidInputError("It doesn't support creating a changefeed with " +
			"the same TiDB cluster as both the source and the target for the changefeed.")
	}

	return nil
}

const numOfAddressParts = 2

func parseMySQLEndpoint(endpoint string) (host string, port int32, err error) {
	parts := strings.Split(endpoint, ":")
	if len(parts) != numOfAddressParts {
		return "", 0, bizerr_wrapper.NewInvalidInputError(
			"MySQL endpoint should be formatted as {host}:{port}")
	}
	uint64Value, err := strconv.ParseUint(strings.TrimSpace(parts[1]), 10, 32)
	if err != nil {
		return "", 0, bizerr_wrapper.NewInvalidInputError(
			"MySQL endpoint's port(%s) is not valid", parts[1])
	}
	return strings.TrimSpace(parts[0]), int32(uint64Value), nil
}

func isOneOfAccessPoints(accessPoints []*core.AccessPoint, mysqlHost string, mysqlPort int32) bool {
	for _, v := range accessPoints {
		if v.PublicEndpoint != nil &&
			addressEqual(v.PublicEndpoint.Host, v.PublicEndpoint.Port, mysqlHost, mysqlPort) {
			return true
		}
		if v.PrivateEndpoint != nil &&
			addressEqual(v.PrivateEndpoint.Host, v.PrivateEndpoint.Port, mysqlHost, mysqlPort) {
			return true
		}
		if v.VpcPeeringEndpoint != nil &&
			addressEqual(v.VpcPeeringEndpoint.Host, v.VpcPeeringEndpoint.Port, mysqlHost, mysqlPort) {
			return true
		}
	}
	return false
}

func addressEqual(tidbHost string, tidbPort int32, mysqlHost string, mysqlPort int32) bool {
	return strings.TrimSpace(tidbHost) == strings.TrimSpace(mysqlHost) && tidbPort == mysqlPort
}