package inner

import (
	"context"

	"github.com/tidbcloud/dataflow-service-ng/app/controllers/console"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"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

	bizChangefeedModel        biz_changefeed.BizChangefeedModel
	ticdcClusterModel         ticdc.ClusterModel
	ticdcChangefeedModel      ticdc.ChangefeedModel
	ticdcClusterConfigManager config_manager.TiCDCClusterConfigManager
	changefeedConfigManager   config_manager.ChangefeedConfigManager
	mgmtCli                   grpcli.MgmtSvcGrpcClient
	bizChangefeedController   console.ChangefeedController
}

func NewOpsInnerController(
	bizChangefeedModel biz_changefeed.BizChangefeedModel,
	ticdcClusterModel ticdc.ClusterModel,
	ticdcChangefeedModel ticdc.ChangefeedModel,
	ticdcClusterConfigManager config_manager.TiCDCClusterConfigManager,
	changefeedConfigManager config_manager.ChangefeedConfigManager,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	bizChangefeedController console.ChangefeedController,
) OpsInnerController {
	return &opsInnerControllerImpl{
		bizChangefeedModel:        bizChangefeedModel,
		ticdcClusterModel:         ticdcClusterModel,
		ticdcChangefeedModel:      ticdcChangefeedModel,
		ticdcClusterConfigManager: ticdcClusterConfigManager,
		changefeedConfigManager:   changefeedConfigManager,
		mgmtCli:                   mgmtCli,
		bizChangefeedController:   bizChangefeedController,
	}
}

func (c *opsInnerControllerImpl) ApplyConfigChanges(
	ctx context.Context, req *dataflow.ApplyConfigChangesReq,
) (*dataflow.ApplyConfigChangesResp, error) {
	ctx = log.WithTraceID(ctx)
	switch req.ClusterType {
	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) 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{
			KernelChangefeedIDs: []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) {
	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 {
			log.GetLogger(ctx).Error("resume changefeed for applying config error", zap.Error(resumeErr))
			if err == nil {
				err = resumeErr.Chain("resume changefeed")
			}
		}
	}()

	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 err := c.bizChangefeedModel.EditBizChangefeedForOps(ctx, changefeedEnt.BizChangefeedID.Uint64()); err != nil {
		return err.Chain("update biz changefeed")
	}
	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
	}

	cluster, err := c.mgmtCli.GetClusterByID(ctx, req.TidbClusterId)
	if err != nil {
		return nil, err.Chain("get tidb cluster")
	}
	ticdcClustersResp, err := c.ticdcClusterModel.List(ctx, &domain.ListTiCDCClustersReq{
		SharedPoolID: utils.NewCluster(cluster).GetSharedPoolId(),
	})
	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)
		bizChangefeed, ok := bizChangefeeds[item.BizChangefeedID.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.BizChangefeedID.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{
			KernelChangefeedIDs: []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)
	bizChangefeedResp, err := c.bizChangefeedModel.GetBizChangefeed(ctx, &domain.GetBizChangefeedReq{
		BizChangefeedID: changefeedEnt.BizChangefeedID.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)
	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{
			KernelChangefeedIDs: []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
}

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{
			KernelChangefeedIDs: []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 {
		return nil, err.Chain("pause changefeed")
	}
	return &emptypb.Empty{}, nil
}

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{
			KernelChangefeedIDs: []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 _, berr := c.ticdcChangefeedModel.ResumeChangefeed(ctx, resumeReq); berr != nil {
		return nil, berr.Chain("resume changefeed")
	}
	return &emptypb.Empty{}, 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),
	}
	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,
		SharedPoolId:      ticdcClusterEnt.SharedPoolID,
		Region:            ticdcClusterEnt.Region,
		Version:           ticdcClusterEnt.Version,
		ProviderType:      string(ticdcClusterEnt.CloudProvider),
		SpecificationCode: ticdcClusterEnt.SpecificationCode,
		Status:            string(ticdcClusterEnt.Status),
		CreatedAt:         timestamppb.New(ticdcClusterEnt.CreatedAt),
	}
}

func (c *opsInnerControllerImpl) dataflowConnector2Internal(
	connector *dataflow.Changefeed,
) *dataflow.InternalChangefeed {
	return &dataflow.InternalChangefeed{
		Id:             connector.ChangefeedId,
		DisplayId:      connector.ChangefeedId,
		DisplayName:    connector.Name,
		DownstreamType: string(convertCDCDownstreamType(connector.ChangefeedType)),
		Status:         string(convertConnectorStatus(connector.Status)),
		CreatedAt:      connector.CreatedAt,
	}
}

func convertCDCDownstreamType(downstreamType dataflow.ChangefeedType) entity.CDCDownStreamType {
	switch downstreamType {
	case dataflow.ChangefeedType_KAFKA:
		return entity.CDCDownStreamTypeKafka
	case dataflow.ChangefeedType_MYSQL:
		return entity.CDCDownStreamTypeMysql
	case dataflow.ChangefeedType_SERVERLESS_TIDB:
		return entity.CDCDownStreamTypeServerlessTiDB
	case dataflow.ChangefeedType_S3:
		return entity.CDCDownStreamTypeS3
	case dataflow.ChangefeedType_GCS:
		return entity.CDCDownStreamTypeGCS
	default:
		return entity.CDCDownStreamTypeAll
	}
}

func convertConnectorStatus(connectorStatus dataflow.Changefeed_Status) entity.CDCStatus {
	switch connectorStatus {
	case dataflow.Changefeed_CREATING:
		return entity.CDCStatusCreating
	case dataflow.Changefeed_RUNNING:
		return entity.CDCStatusNormal
	case dataflow.Changefeed_WARNING:
		return entity.CDCStatusError
	case dataflow.Changefeed_FAILED:
		return entity.CDCStatusFailed
	case dataflow.Changefeed_PAUSING:
		return entity.CDCStatusStopping
	case dataflow.Changefeed_PAUSED:
		return entity.CDCStatusStopped
	case dataflow.Changefeed_RESUMING:
		return entity.CDCStatusStarting
	case dataflow.Changefeed_EDITING:
		return entity.CDCStatusEditing
	case dataflow.Changefeed_DELETING:
		return entity.CDCStatusDeleting
	case dataflow.Changefeed_DELETED:
		return entity.CDCStatusDeleted
	case dataflow.Changefeed_SCALING:
		return entity.CDCStatusScaling
	default:
		return ""
	}
}

//nolint:dupl
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")
	}
	bizChangefeedsResp, err := c.bizChangefeedModel.ListBizChangefeeds(ctx, &domain.ListBizChangefeedsReq{
		PerPage:   uint32(listAllPerPage),
		ClusterID: req.ClusterId,
	})
	if err != nil {
		logger.Error("list biz changefeeds error", zap.Error(err))
		return nil, err.Chain("list biz changefeeds, cluster_id: %d", req.ClusterId)
	}
	if len(bizChangefeedsResp.BizChangefeeds) == 0 {
		logger.Info("no biz changefeeds found, skip pause cluster changefeeds")
		return &emptypb.Empty{}, nil
	}

	for _, cf := range bizChangefeedsResp.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 nil, 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))
	}

	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (c *opsInnerControllerImpl) ResumeClusterChangefeeds(
	ctx context.Context,
	req *dataflow.ResumeClusterChangefeedsRequest,
) (*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")
	}
	bizChangefeedsResp, err := c.bizChangefeedModel.ListBizChangefeeds(ctx, &domain.ListBizChangefeedsReq{
		PerPage:   uint32(listAllPerPage),
		ClusterID: req.ClusterId,
	})
	if err != nil {
		logger.Error("list biz changefeeds error", zap.Error(err))
		return nil, err.Chain("list biz changefeeds, cluster_id: %d", req.ClusterId)
	}
	if len(bizChangefeedsResp.BizChangefeeds) == 0 {
		logger.Info("no biz changefeeds found, skip resume cluster changefeeds")
		return &emptypb.Empty{}, nil
	}

	for _, cf := range bizChangefeedsResp.BizChangefeeds {
		if cf.ChangefeedStatus == entity.ChangefeedKernelStatusNormal ||
			cf.ChangefeedStatus == entity.ChangefeedKernelStatusWarning {
			logger.Info("skip resumed biz changefeed for the status",
				zap.Any("status", cf.ChangefeedStatus),
				zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
				zap.String("name", cf.Name))
			continue
		}
		resumeReq := &domain.ResumeBizChangefeedReq{
			BizChangefeedID: cf.BizChangefeedID,
		}
		if err := c.bizChangefeedModel.ResumeBizChangefeed(ctx, resumeReq); err != nil {
			logger.Error("resume biz changefeed error",
				zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
				zap.String("name", cf.Name),
				zap.Error(err),
			)
			return nil, err.Chain("resume biz changefeed:%d, name: %s", cf.BizChangefeedID, cf.Name)
		}
		logger.Info("resumed biz changefeed successfully",
			zap.Uint64("biz_changefeed_id", cf.BizChangefeedID),
			zap.String("name", cf.Name))
	}

	return &emptypb.Empty{}, nil
}

//nolint:funlen
func (c *opsInnerControllerImpl) UpdateTableFilterOfClusterChangefeeds(
	ctx context.Context,
	req *dataflow.UpdateTableFilterOfClusterChangefeedsRequest,
) (*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 len(req.Updates) == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "updates is required")
	}

	for _, update := range req.Updates {
		if update.ChangefeedId == 0 || len(update.TableFilter) == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid changefeed_id is required")
		}
		if len(update.TableFilter) == 0 {
			return nil, bizerr.New(bizerr.ErrCodeParams, "table_filter is required")
		}
		changefeed, err := c.bizChangefeedController.GetChangefeed(ctx, &dataflow.ChangefeedID{
			ChangefeedId: update.ChangefeedId,
			ClusterId:    req.ClusterId,
		})
		if err != nil {
			logger.Error("get changefeed error", zap.Error(err), zap.Uint64("changefeed_id", update.ChangefeedId))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "get changefeed: %d", update.ChangefeedId)
		}

		// update table filter for different downstream types
		switch changefeed.ChangefeedType {
		case dataflow.ChangefeedType_KAFKA:
			changefeed.Request.KafkaInfo.TableConfig.FilterRule = update.TableFilter
		case dataflow.ChangefeedType_MYSQL:
			changefeed.Request.MysqlInfo.TableConfig.FilterRule = update.TableFilter
		case dataflow.ChangefeedType_SERVERLESS_TIDB:
			changefeed.Request.ServerlessTidbInfo.MysqlInfo.TableConfig.FilterRule = update.TableFilter
		case dataflow.ChangefeedType_S3:
			changefeed.Request.S3Info.TableConfig.FilterRule = update.TableFilter
		case dataflow.ChangefeedType_GCS:
			changefeed.Request.GcsInfo.TableConfig.FilterRule = update.TableFilter
		case dataflow.ChangefeedType_PULSAR:
			changefeed.Request.PulsarInfo.TableConfig.FilterRule = update.TableFilter
		default:
			return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported downstream type: %s", changefeed.ChangefeedType)
		}

		// update table filter for the changefeed
		updateReq := &dataflow.ChangefeedEditReq{
			ClusterId:          req.ClusterId,
			ChangefeedId:       update.ChangefeedId,
			ChangefeedType:     changefeed.ChangefeedType,
			Name:               changefeed.Name,
			KafkaInfo:          changefeed.Request.KafkaInfo,
			MysqlInfo:          changefeed.Request.MysqlInfo,
			ServerlessTidbInfo: changefeed.Request.ServerlessTidbInfo,
			S3Info:             changefeed.Request.S3Info,
			GcsInfo:            changefeed.Request.GcsInfo,
			PulsarInfo:         changefeed.Request.PulsarInfo,
		}
		if _, err := c.bizChangefeedController.EditChangefeed(ctx, updateReq); err != nil {
			logger.Error("edit changefeed error", zap.Error(err), zap.Uint64("changefeed_id", update.ChangefeedId))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "edit changefeed: %d, error: %s", update.ChangefeedId, err.Error())
		}
	}
	logger.Info("updated table filter of cluster changefeeds successfully")
	return &emptypb.Empty{}, nil
}