package inner

import (
	"context"

	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"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/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const listAllPerPage = 1000

type OpsInnerController interface {
	dataflow.OpsInternalServiceServer
}

type opsInnerControllerImpl struct {
	dataflow.UnimplementedOpsInternalServiceServer

	configManager             config_manager.ConfigCenterManager
	bizChangefeedModel        biz_changefeed.BizChangefeedModel
	ticdcClusterModel         ticdc.ClusterModel
	ticdcChangefeedModel      ticdc.ChangefeedModel
	ticdcClusterConfigManager config_manager.TiCDCClusterConfigManager
	changefeedConfigManager   config_manager.ChangefeedConfigManager
	ticdcClusterCli           regionalapi.TiCDCClusterClient
}

func NewOpsInnerController(
	configManager config_manager.ConfigCenterManager,
	bizChangefeedModel biz_changefeed.BizChangefeedModel,
	ticdcClusterModel ticdc.ClusterModel,
	ticdcChangefeedModel ticdc.ChangefeedModel,
	ticdcClusterConfigManager config_manager.TiCDCClusterConfigManager,
	changefeedConfigManager config_manager.ChangefeedConfigManager,
	ticdcClusterCli regionalapi.TiCDCClusterClient,
) OpsInnerController {
	return &opsInnerControllerImpl{
		configManager:             configManager,
		bizChangefeedModel:        bizChangefeedModel,
		ticdcClusterModel:         ticdcClusterModel,
		ticdcChangefeedModel:      ticdcChangefeedModel,
		ticdcClusterConfigManager: ticdcClusterConfigManager,
		changefeedConfigManager:   changefeedConfigManager,
		ticdcClusterCli:           ticdcClusterCli,
	}
}

func (c *opsInnerControllerImpl) ApplyConfigChanges(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) (*dataflow.ApplyConfigChangesResp, error) {
	ctx = log.WithTraceID(ctx)
	switch req.ClusterType {
	case dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER:
		// backup, restore and old ticdc cluster configuration belongs to tidb cluster type.
		// use old logic to apply their configuration changes.
		if err := c.applyTiDBClusterTypeConfigChanges(ctx, req); err != nil {
			log.GetLogger(ctx).Error("apply config changes of tidb cluster type error",
				zap.Any("req", req), zap.Error(err))
			return nil, err
		}
	case dataflow.ConfigClusterType_TICDC_CLUSTER:
		if err := c.applyTiCDCClusterConfig(ctx, req); err != nil {
			log.GetLogger(ctx).Error("apply config changes of ticdc cluster type error",
				zap.Any("req", req), zap.Error(err))
			return nil, err
		}
	case dataflow.ConfigClusterType_CHANGEFEED_CLUSTER:
		if err := c.applyChangefeedConfig(ctx, req); err != nil {
			log.GetLogger(ctx).Error("apply config changes of changefeed type error",
				zap.Any("req", req), zap.Error(err))
			return nil, err
		}
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported cluster type %s", req.ClusterType.String())
	}
	return &dataflow.ApplyConfigChangesResp{}, nil
}

func (c *opsInnerControllerImpl) applyTiDBClusterTypeConfigChanges(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) bizerr.BizError {
	component, bizErr := domain.DataflowConfigFile2Component(req.ConfigurationFile)
	if bizErr != nil {
		return bizErr.Chain("convert file to component")
	}

	updateConfigVersionReq := &domain.UpdateConfigVersionReq{
		ClusterID:      req.ClusterId,
		Component:      component,
		Version:        req.Version,
		TemplateID:     req.TemplateId,
		CustomParams:   req.CustomParams,
		CustomImageTag: req.CustomImageTag,
	}
	if bizErr := c.configManager.UpdateConfigVersion(ctx, updateConfigVersionReq); bizErr != nil {
		return bizErr.Chain("update config and version")
	}
	return nil
}

func (c *opsInnerControllerImpl) applyTiCDCClusterConfig(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) bizerr.BizError {
	updateConfigReq := &domain.UpdateTiCDCClusterConfigReq{
		TiCDCClusterID:     req.ClusterId,
		CustomParams:       req.CustomParams,
		TiDBClusterVersion: req.Version,
		TemplateID:         req.TemplateId,
		CustomImageTag:     req.CustomImageTag,
	}
	if err := c.ticdcClusterConfigManager.Update(ctx, updateConfigReq); err != nil {
		return err.Chain("update config center")
	}

	applyConfigReq := &domain.ApplyTiCDCClusterConfigReq{
		TiCDCClusterID: req.ClusterId,
	}
	if err := c.ticdcClusterModel.ApplyConfig(ctx, applyConfigReq); err != nil {
		return err.Chain("apply ticdc cluster config")
	}
	return nil
}

func (c *opsInnerControllerImpl) applyChangefeedConfig(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) bizerr.BizError {
	queryResp, err := c.ticdcChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			ChangefeedIDs: []uint64{req.ClusterId},
		},
	})
	if err != nil {
		return err.Chain("get ticdc changefeed")
	}
	if err := c.applyTiCDCChangefeedConfig(ctx, req, &queryResp.ChangefeedEntity); err != nil {
		return err.Chain("apply ticdc changefeed config")
	}
	return nil
}

func (c *opsInnerControllerImpl) updateChangefeedConfig(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) bizerr.BizError {
	updateConfigReq := &domain.UpdateChangefeedConfigReq{
		ChangefeedID: req.ClusterId,
	}
	if req.ConfigurationFile == dataflow.ConfigFileType_CHANGEFEED_FILE {
		updateConfigReq.Configuration = req.CustomParams
	} else if req.ConfigurationFile == dataflow.ConfigFileType_SINK_URI {
		updateConfigReq.URIConfiguration = req.CustomParams
	} else {
		return bizerr.New(bizerr.ErrCodeParams, "invalid configuration file")
	}
	if err := c.changefeedConfigManager.Update(ctx, updateConfigReq); err != nil {
		return err.Chain("update config center")
	}
	return nil
}

func (c *opsInnerControllerImpl) applyTiCDCChangefeedConfig(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq, changefeedEnt *entity.TiCDCChangefeed,
) (err bizerr.BizError) {
	lg := log.GetLogger(ctx).With(
		zap.Uint64("cdcClusterId", changefeedEnt.TiCDCClusterID.Uint64()),
		zap.Uint64("cdcChangefeedId", changefeedEnt.ID.Uint64()))

	//nolint:nestif
	if changefeedEnt.KernelMetrics.FeedState != entity.ChangefeedKernelStatusStopped {
		pauseReq := &domain.PauseChangefeedReq{
			ChangefeedID: changefeedEnt.ID.Uint64(),
		}
		if _, err := c.ticdcChangefeedModel.PauseChangefeed(ctx, pauseReq); err != nil {
			return err.Chain("pause changefeed")
		}
		defer func() {
			resumeReq := &domain.ResumeChangefeedReq{
				ChangefeedID: changefeedEnt.ID.Uint64(),
			}
			_, resumeErr := c.ticdcChangefeedModel.ResumeChangefeed(ctx, resumeReq)
			if resumeErr != nil {
				lg.Error("resume changefeed for applying config error", zap.Error(resumeErr))
				if err == nil {
					err = resumeErr.Chain("resume changefeed")
				}
			}
		}()
	} else {
		lg.Info("skip pausing/resuming changefeed because it is already stopped",
			zap.String("state", string(changefeedEnt.KernelMetrics.FeedState)))
	}

	if err := c.updateChangefeedConfig(ctx, req); err != nil {
		return err
	}
	if err := c.updateTiCDCChangefeed(ctx, changefeedEnt); err != nil {
		return err.Chain("update changefeed config")
	}
	return nil
}

func (c *opsInnerControllerImpl) updateTiCDCChangefeed(
	ctx context.Context, changefeedEnt *entity.TiCDCChangefeed,
) bizerr.BizError {
	if changefeedEnt.BusinessType == entity.TiCDCClusterBizTypeChangefeed {
		// for biz changefeed
		if err := c.bizChangefeedModel.EditBizChangefeedForOps(ctx, changefeedEnt.BusinessID.Uint64()); err != nil {
			return err.Chain("update biz changefeed")
		}
		return nil
	}

	// for changefeed of recovery group
	if changefeedEnt.DownStreamType != entity.TiCDCDownStreamTypeMySQL {
		return bizerr.New(bizerr.ErrCodeUnknown,
			"not support changefeed downstream %s", string(changefeedEnt.DownStreamType))
	}
	modifyConfigReq := &domain.ModifyChangefeedConfigReq{
		ChangefeedID: changefeedEnt.ID.Uint64(),
		Config: &domain.TiCDCChangefeedConfig{
			Config:    &domain.TiCDCChangefeedConfigData{ReplicaConfig: &clusterproxy.ReplicaConfig{}},
			URIConfig: &domain.TiCDCChangefeedMySQLURIConfigData{},
		},
	}
	if _, err := c.ticdcChangefeedModel.ModifyChangefeedConfig(ctx, modifyConfigReq); err != nil {
		return err.Chain("modify changefeed config")
	}
	return nil
}

func (c *opsInnerControllerImpl) UpdateTiCDCClusterStorage(
	ctx context.Context, req *dataflow.UpdateTiCDCClusterStorageReq,
) (*dataflow.UpdateTiCDCClusterStorageResp, error) {
	ctx = log.WithTraceID(ctx)
	updateReq := &domain.UpdateTiCDCClusterStorageReq{
		TiCDCClusterID: req.ClusterId,
		SizeGi:         req.SizeGi,
		Iops:           req.Iops,
		ThroughputMbs:  req.ThroughputMbs,
	}
	if err := c.ticdcClusterModel.UpdateStorage(ctx, updateReq); err != nil {
		return nil, err.Chain("failed ticdc cluster storage")
	}
	log.GetLogger(ctx).Info("update ticdc cluster storage success",
		zap.Uint64("cluster_id", req.ClusterId),
		zap.Uint32("size_gi", req.SizeGi),
		zap.Uint32("iops", req.Iops),
		zap.Uint32("throughput_mbs", req.ThroughputMbs))
	return &dataflow.UpdateTiCDCClusterStorageResp{}, nil
}

func (c *opsInnerControllerImpl) ListInternalChangefeeds(
	ctx context.Context, req *dataflow.ListInternalChangefeedsRequest,
) (*dataflow.ListInternalChangefeedsResponse, error) {
	ctx = log.WithTraceID(ctx)
	ticdcChangefeeds, err := c.listTiCDCChangefeeds(ctx, req)
	if err != nil {
		log.GetLogger(ctx).Warn("list ticdc changefeeds error",
			zap.Uint64("tidb_cluster_id", req.TidbClusterId), zap.Error(err))
		return nil, err
	}
	resp := &dataflow.ListInternalChangefeedsResponse{
		Changefeeds: ticdcChangefeeds,
	}
	return resp, nil
}

func (c *opsInnerControllerImpl) listTiCDCChangefeeds(
	ctx context.Context, req *dataflow.ListInternalChangefeedsRequest,
) ([]*dataflow.InternalChangefeed, error) {
	bizChangefeedsResp, err := c.bizChangefeedModel.ListBizChangefeeds(ctx, &domain.ListBizChangefeedsReq{
		PerPage:   uint32(listAllPerPage),
		ClusterID: req.TidbClusterId,
	})
	if err != nil {
		return nil, err.Chain("list biz changefeeds")
	}
	if len(bizChangefeedsResp.BizChangefeeds) == 0 {
		return []*dataflow.InternalChangefeed{}, nil
	}
	bizChangefeeds := make(map[uint64]*domain.BizChangefeed, len(bizChangefeedsResp.BizChangefeeds))
	for _, item := range bizChangefeedsResp.BizChangefeeds {
		bizChangefeeds[item.BizChangefeedID] = item
	}

	ticdcClustersResp, err := c.ticdcClusterModel.List(ctx, &domain.ListTiCDCClustersReq{
		TiDBClusterID: req.TidbClusterId,
	})
	if err != nil {
		return nil, err.Chain("list ticdc clusters")
	}
	if len(ticdcClustersResp.Items) == 0 {
		return []*dataflow.InternalChangefeed{}, nil
	}
	ticdcClusterIDs := make([]uint64, 0, len(ticdcClustersResp.Items))
	internalTiCDCClusters := make(map[uint64]*dataflow.InternalTiCDCCluster, len(ticdcClustersResp.Items))
	for i := range ticdcClustersResp.Items {
		item := &ticdcClustersResp.Items[i]
		ticdcClusterIDs = append(ticdcClusterIDs, item.ID.Uint64())
		internalTiCDCClusters[item.ID.Uint64()] = c.ticdcClusterEntity2DataflowInternal(item)
	}

	return c.assembleTiCDCChangefeeds(ctx, ticdcClusterIDs, internalTiCDCClusters, bizChangefeeds)
}

func (c *opsInnerControllerImpl) assembleTiCDCChangefeeds(
	ctx context.Context,
	ticdcClusterIDs []uint64,
	internalTiCDCClusters map[uint64]*dataflow.InternalTiCDCCluster,
	bizChangefeeds map[uint64]*domain.BizChangefeed,
) ([]*dataflow.InternalChangefeed, error) {
	changefeedsResp, err := c.ticdcChangefeedModel.ListChangefeeds(ctx, &domain.ListChangefeedsReq{
		Filter: &domain.ListChangefeedFilter{
			TiCDCClusterIDs: ticdcClusterIDs,
			Deleted:         false,
		},
		Pager: &domain.Pager{
			PageSize: int64(listAllPerPage),
		},
	})
	if err != nil {
		return nil, err.Chain("list ticdc changefeeds")
	}

	internalChangefeeds := make([]*dataflow.InternalChangefeed, 0, len(changefeedsResp.Items))
	for i := range changefeedsResp.Items {
		item := &changefeedsResp.Items[i].ChangefeedEntity
		internalChangefeed := c.ticdcChangefeedEntity2DataflowInternal(item)
		if item.BusinessType == entity.TiCDCClusterBizTypeChangefeed {
			bizChangefeed, ok := bizChangefeeds[item.BusinessID.Uint64()]
			if !ok {
				// biz changefeed may be deleting or deleted, filter abnormal kernel changefeed
				log.GetLogger(ctx).Info(
					"biz changefeed is not found, skip the kernel changefeed",
					zap.Uint64("ticdc_changefeed_id", item.ID.Uint64()),
					zap.Uint64("biz_changefeed_id", item.BusinessID.Uint64()))
				continue
			}
			internalChangefeed.DisplayId = bizChangefeed.BizChangefeedID
			internalChangefeed.DisplayName = bizChangefeed.Name
		}
		if internalChangefeed.TicdcCluster == nil {
			internalTiCDCCluster, ok := internalTiCDCClusters[item.TiCDCClusterID.Uint64()]
			if !ok {
				log.GetLogger(ctx).Info(
					"ticdc cluster is not found, skip the kernel changefeed",
					zap.Uint64("ticdc_changefeed_id", item.ID.Uint64()),
					zap.Uint64("ticdc_cluster_id", item.TiCDCClusterID.Uint64()))
				continue
			}
			internalChangefeed.TicdcCluster = internalTiCDCCluster
		}
		internalChangefeeds = append(internalChangefeeds, internalChangefeed)
	}
	return internalChangefeeds, nil
}

func (c *opsInnerControllerImpl) GetInternalChangefeed(
	ctx context.Context, req *dataflow.GetInternalChangefeedRequest,
) (*dataflow.InternalChangefeed, error) {
	ctx = log.WithTraceID(ctx)
	internalChangefeed, bizErr := c.getTiCDCChangefeed(ctx, req.ChangefeedId)
	if bizErr != nil {
		log.GetLogger(ctx).Warn("get ticdc changefeed error",
			zap.Uint64("changefeed_id", req.ChangefeedId), zap.Error(bizErr))
		return nil, bizErr
	}
	return internalChangefeed, nil
}

func (c *opsInnerControllerImpl) getTiCDCChangefeed(
	ctx context.Context, changefeedID uint64,
) (*dataflow.InternalChangefeed, bizerr.BizError) {
	queryResp, err := c.ticdcChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			ChangefeedIDs: []uint64{changefeedID},
		},
	})
	if err != nil {
		return nil, err.Chain("get ticdc changefeed")
	}

	changefeedEnt := &queryResp.ChangefeedEntity
	if changefeedEnt.TiCDCCluster == nil {
		return nil, bizerr.New(
			bizerr.ErrCodeUnknown, "ticdc cluster %d is not found", changefeedEnt.TiCDCClusterID.Uint64())
	}

	internalChangefeed := c.ticdcChangefeedEntity2DataflowInternal(changefeedEnt)
	if changefeedEnt.BusinessType == entity.TiCDCClusterBizTypeChangefeed {
		bizChangefeedResp, err := c.bizChangefeedModel.GetBizChangefeed(ctx, &domain.GetBizChangefeedReq{
			BizChangefeedID: changefeedEnt.BusinessID.Uint64(),
		})
		if err != nil {
			return nil, err.Chain("get biz changefeed")
		}
		internalChangefeed.DisplayId = bizChangefeedResp.BizChangefeed.BizChangefeedID
		internalChangefeed.DisplayName = bizChangefeedResp.BizChangefeed.Name
	}
	return internalChangefeed, nil
}

func (c *opsInnerControllerImpl) GetInternalTiCDCCluster(
	ctx context.Context, req *dataflow.GetInternalTiCDCClusterRequest,
) (*dataflow.InternalTiCDCCluster, error) {
	ctx = log.WithTraceID(ctx)
	queryResp, err := c.ticdcClusterModel.Query(ctx, &domain.QueryTiCDCClusterReq{
		ID: req.TicdcClusterId,
	})
	if err != nil {
		log.GetLogger(ctx).Warn("get ticdc clster error",
			zap.Uint64("ticdc_cluster_id", req.TicdcClusterId), zap.Error(err))
		return nil, err.Chain("get ticdc cluster")
	}
	internalTiCDCCluster := c.ticdcClusterEntity2DataflowInternal(queryResp.TiCDCClusterEntity)
	if err := c.addInternalDeploymentInfo(ctx, queryResp.TiCDCClusterEntity, internalTiCDCCluster); err != nil {
		log.GetLogger(ctx).Warn("add internal deployment info error",
			zap.Uint64("ticdc_cluster_id", req.TicdcClusterId), zap.Error(err))
		return nil, err.Chain("add internal deployment info")
	}
	return internalTiCDCCluster, nil
}

func (c *opsInnerControllerImpl) RestartTiCDCCluster(
	ctx context.Context, req *dataflow.RestartTiCDCClusterRequest,
) (*emptypb.Empty, error) {
	restartReq := &domain.RestartTiCDCClusterReq{
		TiCDCClusterID: req.TicdcClusterId,
	}
	if err := c.ticdcClusterModel.Restart(ctx, restartReq); err != nil {
		log.GetLogger(ctx).Warn("restart ticdc cluster error",
			zap.Uint64("ticdc_cluster_id", req.TicdcClusterId), zap.Error(err))
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (c *opsInnerControllerImpl) RestartChangefeed(
	ctx context.Context, req *dataflow.RestartChangefeedRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Uint64("changefeed_id", req.ChangefeedId))
	_, err := c.ticdcChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			ChangefeedIDs: []uint64{req.ChangefeedId},
		},
	})
	if err != nil {
		logger.Warn("get ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("get ticdc changefeed")
	}
	if err := c.restartTiCDCChangefeed(ctx, req.ChangefeedId); err != nil {
		logger.Error("restart ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("restart ticdc changefeed")
	}
	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (c *opsInnerControllerImpl) PauseChangefeed(
	ctx context.Context, req *dataflow.PauseChangefeedRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Uint64("changefeed_id", req.ChangefeedId))

	_, err := c.ticdcChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			ChangefeedIDs: []uint64{req.ChangefeedId},
		},
	})
	if err != nil {
		logger.Warn("get ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("get ticdc changefeed")
	}
	pauseReq := &domain.PauseChangefeedReq{
		ChangefeedID: req.ChangefeedId,
	}
	if _, err := c.ticdcChangefeedModel.PauseChangefeed(ctx, pauseReq); err != nil {
		logger.Error("pause ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("pause ticdc changefeed")
	}
	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (c *opsInnerControllerImpl) ResumeChangefeed(
	ctx context.Context, req *dataflow.ResumeChangefeedRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Uint64("changefeed_id", req.ChangefeedId))

	_, err := c.ticdcChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			ChangefeedIDs: []uint64{req.ChangefeedId},
		},
	})
	if err != nil {
		logger.Warn("get ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("get ticdc changefeed")
	}
	resumeReq := &domain.ResumeChangefeedReq{
		ChangefeedID: req.ChangefeedId,
	}
	if _, err := c.ticdcChangefeedModel.ResumeChangefeed(ctx, resumeReq); err != nil {
		logger.Error("resume ticdc changefeed error", zap.Error(err))
		return nil, err.Chain("resume ticdc changefeed")
	}
	return &emptypb.Empty{}, nil
}

func (c *opsInnerControllerImpl) PauseClusterChangefeeds(
	ctx context.Context, req *dataflow.PauseClusterChangefeedsRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.Uint64("cluster_id", req.ClusterId))
	if req.ClusterId == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	if err := c.pauseClusterBizChangefeeds(ctx, req); err != nil {
		logger.Error("pause cluster biz changefeeds error", zap.Error(err))
		return nil, err.Chain("pause cluster biz changefeeds")
	}
	return &emptypb.Empty{}, nil
}

const changefeedPageSize = uint32(100)

//nolint:funlen
func (c *opsInnerControllerImpl) pauseClusterBizChangefeeds(
	ctx context.Context,
	req *dataflow.PauseClusterChangefeedsRequest,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	page := uint32(1)
	perPage := changefeedPageSize // reasonable page size

	for {
		changefeeds, err := c.bizChangefeedModel.ListBizChangefeeds(ctx, &domain.ListBizChangefeedsReq{
			ClusterID: req.ClusterId,
			Page:      page,
			PerPage:   perPage,
		})
		if err != nil {
			logger.Error("list biz changefeeds error", zap.Uint64("cluster_id", req.ClusterId), zap.Error(err))
			return err.Chain("list biz changefeeds: cluster_id: %d", req.ClusterId)
		}

		if len(changefeeds.BizChangefeeds) == 0 {
			logger.Info("no biz changefeeds found", zap.Uint64("cluster_id", req.ClusterId))
			break
		}

		// Pause each changefeed one by one
		for _, cf := range changefeeds.BizChangefeeds {
			if cf.ChangefeedStatus == entity.ChangefeedKernelStatusFailed ||
				cf.ChangefeedStatus == entity.ChangefeedKernelStatusStopped {
				logger.Info("skip paused biz changefeed for the status",
					zap.Any("status", cf.ChangefeedStatus),
					zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
					zap.String("name", cf.Name))
				continue
			}
			pauseReq := &domain.PauseBizChangefeedReq{
				BizChangefeedID: cf.BizChangefeedID,
			}
			if err := c.bizChangefeedModel.PauseBizChangefeed(ctx, pauseReq); err != nil {
				logger.Error("pause biz changefeed error",
					zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
					zap.String("name", cf.Name),
					zap.Error(err))
				return err.Chain("pause biz changefeed:%d, name: %s", cf.BizChangefeedID, cf.Name)
			}
			logger.Info("paused biz changefeed successfully",
				zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
				zap.String("name", cf.Name))
		}

		// Check if we've processed all changefeeds
		if page*perPage >= changefeeds.Total {
			logger.Info("processed all biz changefeeds",
				zap.Uint64("cluster_id", req.ClusterId),
				zap.Uint32("page", page),
				zap.Uint32("per_page", perPage),
				zap.Uint32("total", changefeeds.Total))
			break
		}
		page++
	}

	return nil
}

func (c *opsInnerControllerImpl) restartTiCDCChangefeed(ctx context.Context, changefeedID uint64) bizerr.BizError {
	pauseReq := &domain.PauseChangefeedReq{
		ChangefeedID: changefeedID,
	}
	if _, err := c.ticdcChangefeedModel.PauseChangefeed(ctx, pauseReq); err != nil {
		return err.Chain("pause changefeed")
	}
	resumeReq := &domain.ResumeChangefeedReq{
		ChangefeedID: changefeedID,
	}
	if _, err := c.ticdcChangefeedModel.ResumeChangefeed(ctx, resumeReq); err != nil {
		return err.Chain("resume changefeed")
	}
	return nil
}

func (c *opsInnerControllerImpl) ticdcChangefeedEntity2DataflowInternal(
	changefeedEnt *entity.TiCDCChangefeed,
) *dataflow.InternalChangefeed {
	internalChangefeed := &dataflow.InternalChangefeed{
		Id:             changefeedEnt.ID.Uint64(),
		DisplayId:      changefeedEnt.ID.Uint64(),
		DisplayName:    changefeedEnt.KernelChangefeedID,
		DownstreamType: string(changefeedEnt.DownStreamType),
		Status:         string(changefeedEnt.KernelMetrics.FeedState),
		CreatedAt:      timestamppb.New(changefeedEnt.CreatedAt),
		OrgId:          changefeedEnt.TenantID.Uint64(),
		ProjectId:      changefeedEnt.ProjectID.Uint64(),
		TidbClusterId:  changefeedEnt.UpStreamClusterID.Uint64(),
	}
	if changefeedEnt.TiCDCCluster != nil {
		internalChangefeed.TicdcCluster = c.ticdcClusterEntity2DataflowInternal(changefeedEnt.TiCDCCluster)
	}
	return internalChangefeed
}

func (c *opsInnerControllerImpl) ticdcClusterEntity2DataflowInternal(
	ticdcClusterEnt *entity.TiCDCCluster,
) *dataflow.InternalTiCDCCluster {
	return &dataflow.InternalTiCDCCluster{
		Id:            ticdcClusterEnt.ID.Uint64(),
		Name:          ticdcClusterEnt.Name,
		TidbClusterId: ticdcClusterEnt.ClusterID.Uint64(),
		Region:        ticdcClusterEnt.Region,
		Version:       ticdcClusterEnt.Version,
		ProviderType: string(domain.InfraProviderType2CloudProviderType(
			domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
		SpecificationCode: ticdcClusterEnt.SpecificationCode,
		Status:            string(ticdcClusterEnt.Status),
		CreatedAt:         timestamppb.New(ticdcClusterEnt.CreatedAt),
		DeployType:        string(ticdcClusterEnt.DeployType),
		OrgId:             ticdcClusterEnt.TenantID.Uint64(),
		ProjectId:         ticdcClusterEnt.ProjectID.Uint64(),
	}
}

func (c *opsInnerControllerImpl) addInternalDeploymentInfo(ctx context.Context,
	ticdcClusterEnt *entity.TiCDCCluster,
	internalCDCCluster *dataflow.InternalTiCDCCluster,
) bizerr.BizError {
	controlplaneArch := "cloud-native"
	if ticdcClusterEnt.IsRegional {
		controlplaneArch = "3layer"
	}

	operatorVersion := "v1"
	if ticdcClusterEnt.IsRegional {
		region := &regionalapi.Region{
			TenantID: ticdcClusterEnt.TenantID.Uint64(),
			Name:     ticdcClusterEnt.Region,
			CloudProvider: string(domain.InfraProviderType2CloudProviderType(
				domain.InfraProviderType(ticdcClusterEnt.InfraProviderType))),
		}
		cluster, err := c.ticdcClusterCli.GetTiCDCCluster(ctx, region, &ticdc_v1.GetTiCDCClusterRequest{
			TicdcClusterId: ticdcClusterEnt.Name,
		})
		if err != nil {
			return err.Chain("get ticdc cluster")
		}
		operatorVersion = cluster.OperatorVersion
	}
	internalCDCCluster.ControlplaneArch = controlplaneArch
	internalCDCCluster.OperatorVersion = operatorVersion
	return nil
}

func (c *opsInnerControllerImpl) dataflowConnector2Internal(
	connector *dataflow.Connector,
) *dataflow.InternalChangefeed {
	return &dataflow.InternalChangefeed{
		Id:             connector.ConnectorId,
		DisplayId:      connector.ConnectorId,
		DisplayName:    connector.Name,
		DownstreamType: string(convertCDCDownstreamType(connector.ConnectorType)),
		Status:         string(convertConnectorStatus(connector.Status)),
		CreatedAt:      connector.CreatedAt,
		OrgId:          connector.Request.OrgId,
		ProjectId:      connector.Request.ProjectId,
		TidbClusterId:  connector.Request.ClusterId,
	}
}

func convertCDCDownstreamType(downstreamType dataflow.ConnectorType) entity.CDCDownStreamType {
	switch downstreamType {
	case dataflow.ConnectorType_KAFKA:
		return entity.CDCDownStreamTypeKafka
	case dataflow.ConnectorType_MYSQL:
		return entity.CDCDownStreamTypeMysql
	case dataflow.ConnectorType_SERVERLESS_TIDB:
		return entity.CDCDownStreamTypeServerlessTiDB
	case dataflow.ConnectorType_S3:
		return entity.CDCDownStreamTypeS3
	case dataflow.ConnectorType_GCS:
		return entity.CDCDownStreamTypeGCS
	default:
		return entity.CDCDownStreamTypeAll
	}
}

func convertConnectorStatus(connectorStatus dataflow.Connector_Status) entity.CDCStatus {
	switch connectorStatus {
	case dataflow.Connector_CREATING:
		return entity.CDCStatusCreating
	case dataflow.Connector_RUNNING:
		return entity.CDCStatusNormal
	case dataflow.Connector_WARNING:
		return entity.CDCStatusError
	case dataflow.Connector_FAILED:
		return entity.CDCStatusFailed
	case dataflow.Connector_PAUSING:
		return entity.CDCStatusStopping
	case dataflow.Connector_PAUSED:
		return entity.CDCStatusStopped
	case dataflow.Connector_RESUMING:
		return entity.CDCStatusStarting
	case dataflow.Connector_EDITING:
		return entity.CDCStatusEditing
	case dataflow.Connector_DELETING:
		return entity.CDCStatusDeleting
	case dataflow.Connector_DELETED:
		return entity.CDCStatusDeleted
	case dataflow.Connector_SCALING:
		return entity.CDCStatusScaling
	default:
		return ""
	}
}