package impl

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strconv"
	"time"

	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/private_link"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc"
	ticdc_impl "github.com/tidbcloud/dataflow-service/app/models/ticdc/impl"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/pkg/billing"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

const bizChangefeedPerClusterQuota = 100

type BizChangefeedModelImpl struct {
	transProvider              db.TransProvider
	bizChangefeedRepo          repo.BizChangefeedRepo
	changefeedInstanceTypeRepo repo.ChangefeedInstanceTypeRepo

	kernelChangefeedModelFactory  *ticdc_impl.ChangefeedModelFactory
	regionalKernelChangefeedModel ticdc.ChangefeedModel
	cdcClusterModel               ticdc.ClusterModel
	machineManager                ticdc.MachineManager
	sinkEndpointModel             private_link.SinkEndpointModel

	importCli      grpcli.ImportGrpcClient
	coreCli        grpcli.CoreGrpcClient
	workflowCli    workflow.Workflow
	billingHandler billing.BillingHandler

	downstreamAdaptorFactory *DownstreamAdaptorFactory
	statusMatrix             status_matrix.StatusMatrix
	hooker                   biz_changefeed.BizChangefeedHooker
	clusterJob               cluster_job.Model
}

func NewBizChangefeedModel(
	transProvider db.TransProvider,
	bizChangefeedRepo repo.BizChangefeedRepo,
	changefeedInstanceTypeRepo repo.ChangefeedInstanceTypeRepo,
	regionalKernelChangefeedModel ticdc.ChangefeedModel,
	cdcClusterModel ticdc.ClusterModel,
	machineManager ticdc.MachineManager,
	sinkEndpointModel private_link.SinkEndpointModel,
	importCli grpcli.ImportGrpcClient,
	coreCli grpcli.CoreGrpcClient,
	workflowCli workflow.Workflow,
	billingHandler billing.BillingHandler,
	downstreamAdaptorFactory *DownstreamAdaptorFactory,
	hooker biz_changefeed.BizChangefeedHooker,
	clusterJob cluster_job.Model,
) biz_changefeed.BizChangefeedModel {
	statusMatrix := NewBizChangefeedStatusMatrix()
	kernelChangefeedModelFactory := ticdc_impl.NewChangefeedModelFactory(
		regionalKernelChangefeedModel)
	return &BizChangefeedModelImpl{
		transProvider:                 transProvider,
		bizChangefeedRepo:             bizChangefeedRepo,
		changefeedInstanceTypeRepo:    changefeedInstanceTypeRepo,
		kernelChangefeedModelFactory:  kernelChangefeedModelFactory,
		regionalKernelChangefeedModel: regionalKernelChangefeedModel,
		cdcClusterModel:               cdcClusterModel,
		machineManager:                machineManager,
		sinkEndpointModel:             sinkEndpointModel,
		importCli:                     importCli,
		coreCli:                       coreCli,
		workflowCli:                   workflowCli,
		billingHandler:                billingHandler,
		downstreamAdaptorFactory:      downstreamAdaptorFactory,
		hooker:                        hooker,
		statusMatrix:                  statusMatrix,
		clusterJob:                    clusterJob,
	}
}

func (m *BizChangefeedModelImpl) GetBizChangefeed(
	ctx context.Context, req *domain.GetBizChangefeedReq,
) (*domain.GetBizChangefeedResp, bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz_changefeed")
	}
	bizChangefeed := bizChangefeedEntity2Domain(bizChangefeedEntity)
	kernelChangefeedEntity, err := m.queryKernelChangefeed(ctx, req.BizChangefeedID)
	if err == nil {
		fillKernelChangefeedInfo(bizChangefeed, kernelChangefeedEntity)
	} else if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, err.Chain("query kernel changefeed")
	}
	return &domain.GetBizChangefeedResp{
		BizChangefeed: bizChangefeed,
	}, nil
}

func (m *BizChangefeedModelImpl) ListBizChangefeeds(
	ctx context.Context, req *domain.ListBizChangefeedsReq,
) (*domain.ListBizChangefeedsResp, bizerr.BizError) {
	filterOp := genListBizChangefeedsFilterOp(req)
	pagerOp := genListBizChangefeedsPagerOp(req)
	total, err := m.bizChangefeedRepo.Count(ctx, filterOp)
	if err != nil {
		return nil, err.Chain("count biz_changefeeds")
	}
	bizChangefeedEntities, err := m.bizChangefeedRepo.List(ctx, filterOp, pagerOp)
	if err != nil {
		return nil, err.Chain("list biz_changefeeds")
	}
	if len(bizChangefeedEntities) == 0 {
		return &domain.ListBizChangefeedsResp{BizChangefeeds: []*domain.BizChangefeed{}}, nil
	}

	bizChangefeedIDs := pie.Map(bizChangefeedEntities, func(ent entity.BizChangefeed) uint64 {
		return ent.ID.Uint64()
	})
	kernelChangefeedEntities, err := m.listKernelChangefeeds(ctx, bizChangefeedIDs)
	if err != nil {
		return nil, err.Chain("list kernel changefeeds")
	}
	kernelChangefeedMap := kernelChangefeedEntities2Map(kernelChangefeedEntities)

	bizChangefeeds := make([]*domain.BizChangefeed, 0, len(bizChangefeedEntities))
	for i := range bizChangefeedEntities {
		bizChangefeed := bizChangefeedEntity2Domain(&bizChangefeedEntities[i])
		if kernelChangefeed, ok := kernelChangefeedMap[bizChangefeed.BizChangefeedID]; ok {
			fillKernelChangefeedInfo(bizChangefeed, kernelChangefeed)
		}
		bizChangefeeds = append(bizChangefeeds, bizChangefeed)
	}
	return &domain.ListBizChangefeedsResp{
		BizChangefeeds: bizChangefeeds,
		Total:          uint32(total),
	}, nil
}

func (m *BizChangefeedModelImpl) CreateBizChangefeed(
	ctx context.Context, req *domain.CreateBizChangefeedReq,
) (resp *domain.CreateBizChangefeedResp, err bizerr.BizError) {
	logger := log.GetLogger(ctx)
	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(req.DownstreamType)
	if err != nil {
		return nil, err.Chain("get downstream adaptor")
	}
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}

	if err = m.preCheckCreateBizChangefeed(ctx, req, cluster, downstreamAdaptor); err != nil {
		return nil, err.Chain("precheck create biz_changefeed")
	}
	if err := downstreamAdaptor.EncryptPasswordInPlace(ctx, req.DownstreamParameter); err != nil {
		return nil, err.Chain("encrypt password")
	}
	if req.PrivateLinkEndpointID == 0 {
		req.PrivateLinkEndpointID = downstreamAdaptor.GetPrivateLinkEndpointID(req.DownstreamParameter)
	}
	bizChangefeedEntity, err := m.genBizChangefeedEntity(ctx, req, cluster)
	if err != nil {
		return nil, err.Chain("generate biz_changefeed entity")
	}
	if req.DryRun {
		logger.Info("dry run create biz_changefeed successfully")
		return &domain.CreateBizChangefeedResp{}, nil
	}

	defer func() { m.handleEventHook(ctx, err, entity.CreateChangefeed, req.UserID, bizChangefeedEntity) }()
	resp = &domain.CreateBizChangefeedResp{}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.bizChangefeedRepo.Create(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("create biz_changefeed entity")
		}
		if err := m.createSinkEndpointUsageIfNeeded(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("create sink endpoint usage")
		}
		dagInsID, err := m.createBizChangefeedWorkflow(
			ctx, bizChangefeedEntity,
			workflow.DagCreateBizChangefeed, string(workflow.BusinessActionCreateBizChangefeed))
		if err != nil {
			return err.Chain("create biz_changefeed workflow")
		}
		resp.BizChangefeedID = bizChangefeedEntity.ID.Uint64()
		resp.WorkflowID = dagInsID
		if err := m.clusterJob.RegisterCDCClusterJob(ctx, cluster.Id,
			cluster_job.GetBizChangefeedTaskID(ctx, bizChangefeedEntity.ID.Uint64())); err != nil {
			return err.Chain("register cluster job")
		}
		return nil
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return resp, nil
}

func (m *BizChangefeedModelImpl) genBizChangefeedEntity(
	ctx context.Context, req *domain.CreateBizChangefeedReq, cluster *core.Cluster,
) (*entity.BizChangefeed, bizerr.BizError) {
	downstreamType, bizErr := convertDownstreamType(req.DownstreamType)
	if bizErr != nil {
		return nil, bizErr.Chain("convert downstream type")
	}
	downstreamParamBytes, err := json.Marshal(req.DownstreamParameter)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "marshal downstream params error: %s", err.Error())
	}
	networkInfo, networkType, bizE := getNetworkInfoStrAndType(req.NetworkInfo)
	if bizE != nil {
		return nil, bizE
	}
	if req.PrivateLinkEndpointID != 0 {
		log.GetLogger(ctx).Info("create biz changefeed with private link",
			zap.Int64("private link endpoint id",
				int64(req.PrivateLinkEndpointID)))
	}
	bizChangefeedEntity := &entity.BizChangefeed{
		ID:                0, // ID will be auto-generated by DB
		Name:              req.Name,
		ClusterID:         entity.ID(cluster.Id),
		ProjectID:         entity.ID(cluster.ProjectId),
		TenantID:          entity.ID(cluster.TenantId),
		DownstreamType:    downstreamType,
		DownstreamParams:  string(downstreamParamBytes),
		SpecificationCode: req.SpecificationCode,
		NetworkType:       networkType,
		NetworkInfo:       networkInfo,
		DeployType:        req.DeployType,
		OperationStatus:   entity.BizChangefeedOperationStatusCreating,
		CreatorID:         entity.ID(req.UserID),
		IsRegional:        true,
		BizPLEndpointID:   req.PrivateLinkEndpointID,
	}
	return bizChangefeedEntity, nil
}

func (m *BizChangefeedModelImpl) createSinkEndpointUsageIfNeeded(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if bizChangefeedEntity.BizPLEndpointID == 0 {
		return nil
	}
	req := &domain.CreateSinkEndpointUsageReq{
		SinkEndpointID: bizChangefeedEntity.BizPLEndpointID.Uint64(),
		BusinessType:   entity.BizPLEndpointUsageBusinessTypeChangefeed,
		BusinessID:     bizChangefeedEntity.ID.Uint64(),
	}
	_, err := m.sinkEndpointModel.CreateSinkEndpointUsage(ctx, req)
	return err
}

func (m *BizChangefeedModelImpl) preCheckCreateBizChangefeed(
	ctx context.Context,
	req *domain.CreateBizChangefeedReq,
	cluster *core.Cluster,
	downstreamAdaptor biz_changefeed.DownstreamAdaptor,
) bizerr.BizError {
	// TODO check version when use private link
	if err := m.checkUpstreamCluster(ctx, cluster); err != nil {
		return err.Chain("check upstream cluster")
	}
	if err := m.checkSpecificationCode(ctx, req.SpecificationCode); err != nil {
		return err.Chain("check specification code")
	}
	if err := m.checkRunningImports(ctx, cluster); err != nil {
		return err.Chain("check running imports")
	}
	if err := m.checkBizChangefeedQuota(ctx, cluster); err != nil {
		return err.Chain("check changefeed quota")
	}
	if err := m.checkDownstream(ctx, req, cluster, downstreamAdaptor); err != nil {
		return err.Chain("check downstream")
	}
	return nil
}

func (m *BizChangefeedModelImpl) checkDownstream(
	ctx context.Context,
	req *domain.CreateBizChangefeedReq,
	cluster *core.Cluster,
	downstreamAdaptor biz_changefeed.DownstreamAdaptor,
) bizerr.BizError {
	if err := downstreamAdaptor.TestConnection(ctx, cluster, req.DownstreamParameter); err != nil {
		return err.Chain("test connection")
	}
	if err := downstreamAdaptor.PreCheckChangefeedParameters(ctx, cluster, req.DownstreamParameter); err != nil {
		return err.Chain("check changefeed parameters")
	}
	return nil
}

func (m *BizChangefeedModelImpl) checkUpstreamCluster(ctx context.Context, cluster *core.Cluster) bizerr.BizError {
	if cluster.IsDevTier {
		return bizerr.New(config.ErrCodeCDCNotallowDevTierErr, config.ErrMsgCDCNotallowDevTierErr)
	}

	if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_RECOVERING {
		return bizerr.New(config.ErrCodeCDCRecoveringErr, config.ErrMsgCDCRecoveringErr)
	} else if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_INITIALIZING ||
		cluster.Status == core.ClusterStatus_CLUSTER_STATUS_PREPARING ||
		cluster.Status == core.ClusterStatus_CLUSTER_STATUS_LAUNCHING ||
		cluster.Status == core.ClusterStatus_CLUSTER_STATUS_RESUMING {
		return bizerr.New(config.ErrCodeCDCCreatingErr, config.ErrMsgCDCCreatingErr)
	} else if cluster.Status == core.ClusterStatus_CLUSTER_STATUS_PAUSED ||
		cluster.Status == core.ClusterStatus_CLUSTER_STATUS_PAUSING {
		return bizerr.New(config.ErrCodeClusterPausing, config.ErrMsgClusterPausing)
	}
	return nil
}

func (m *BizChangefeedModelImpl) checkBizChangefeedQuota(ctx context.Context, cluster *core.Cluster) bizerr.BizError {
	bizChangefeeds, err := m.bizChangefeedRepo.List(ctx,
		func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
			return qs.ClusterIDEq(entity.ID(cluster.Id))
		})
	if err != nil {
		return err.Chain("list changefeeds error")
	}
	if len(bizChangefeeds) >= bizChangefeedPerClusterQuota {
		return bizerr.New(
			config.ErrCodeChangefeedPerClusterExceeded,
			config.ErrMsgChangefeedPerClusterExceeded, len(bizChangefeeds))
	}
	return nil
}

func (m *BizChangefeedModelImpl) checkRunningImports(ctx context.Context, cluster *core.Cluster) bizerr.BizError {
	resp, err := m.importCli.List(ctx, &dataflow.ListImportReq{
		OrgId:     cluster.TenantId,
		ProjectId: cluster.ProjectId,
		ClusterId: cluster.Id,
		Statuses: []dataflow.ImportGetResp_Status{
			dataflow.ImportGetResp_PREPARING,
			dataflow.ImportGetResp_IMPORTING,
			dataflow.ImportGetResp_CANCELING,
		},
	})
	if err != nil {
		return err.Chain("list import tasks")
	}
	if resp != nil && len(resp.Imports) > 0 {
		return bizerr.New(config.ErrCodeCDCImportingErr, config.ErrMsgCDCImportingErr)
	}
	return nil
}

func (m *BizChangefeedModelImpl) ScaleBizChangefeed(
	ctx context.Context, req *domain.ScaleBizChangefeedReq,
) (resp *domain.ScaleBizChangefeedResp, err bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz_changefeed")
	}
	if err = m.preCheckScaleBizChangefeed(ctx, req, bizChangefeedEntity); err != nil {
		return nil, err.Chain("precheck scale biz_changefeed")
	}
	defer func() { m.handleEventHook(ctx, err, entity.ScaleChangefeed, req.UserID, bizChangefeedEntity) }()

	resp = &domain.ScaleBizChangefeedResp{}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.updateBizChangefeedEntityForScale(ctx, req, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		dagInsID, err := m.createBizChangefeedWorkflow(
			ctx, bizChangefeedEntity, workflow.DagScaleBizChangefeed, string(workflow.BusinessActionScaleBizChangefeed))
		if err != nil {
			return err.Chain("create biz_changefeed workflow")
		}
		resp.WorkflowID = dagInsID
		return nil
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return resp, nil
}

func (m *BizChangefeedModelImpl) updateBizChangefeedEntityForScale(
	ctx context.Context, req *domain.ScaleBizChangefeedReq, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	nextStatus, err := m.calBizChangefeedNextStatus(ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionScale)
	if err != nil {
		return err.Chain("calculate next status")
	}

	bizChangefeedEntity.OperationStatus = nextStatus
	bizChangefeedEntity.SpecificationCode = req.SpecificationCode
	if err = m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("update biz_changefeed entity")
	}
	return nil
}

func (m *BizChangefeedModelImpl) preCheckScaleBizChangefeed(
	ctx context.Context, req *domain.ScaleBizChangefeedReq, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if bizChangefeedEntity.SpecificationCode == req.SpecificationCode {
		return bizerr.New(bizerr.ErrCodeParams, "specification code is not changed")
	}
	if err := m.checkBizChangefeedStatus(
		ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionScale); err != nil {
		return err.Chain("check biz_changefeed status")
	}
	if err := m.checkSpecificationCode(ctx, req.SpecificationCode); err != nil {
		return err.Chain("check specification code")
	}
	return nil
}

func (m *BizChangefeedModelImpl) DeleteBizChangefeed(
	ctx context.Context, req *domain.DeleteBizChangefeedReq,
) (resp *domain.DeleteBizChangefeedResp, err bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz_changefeed")
	}
	if err = m.preCheckDeleteBizChangefeed(ctx, bizChangefeedEntity); err != nil {
		return nil, err.Chain("precheck delete biz_changefeed")
	}
	defer func() { m.handleEventHook(ctx, err, entity.DeleteChangefeed, req.UserID, bizChangefeedEntity) }()

	resp = &domain.DeleteBizChangefeedResp{}
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.updateBizChangefeedEntityForDelete(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		if err := m.deleteSinkEndpointUsageIfNeeded(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("delete sink endpoint usage")
		}
		dagInsID, err := m.createBizChangefeedWorkflow(
			ctx, bizChangefeedEntity,
			workflow.DagDeleteBizChangefeed, string(workflow.BusinessActionDeleteBizChangefeed))
		if err != nil {
			return err.Chain("create biz_changefeed workflow")
		}
		resp.WorkflowID = dagInsID
		return nil
	})
	if transErr != nil {
		return nil, convertTransactionError(transErr)
	}
	return resp, nil
}

func (m *BizChangefeedModelImpl) updateBizChangefeedEntityForDelete(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	nextStatus, err := m.calBizChangefeedNextStatus(ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionDelete)
	if err != nil {
		return err.Chain("calculate next status")
	}

	bizChangefeedEntity.OperationStatus = nextStatus
	if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("update biz_changefeed entity")
	}
	return nil
}

func (m *BizChangefeedModelImpl) deleteSinkEndpointUsageIfNeeded(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if bizChangefeedEntity.BizPLEndpointID == 0 {
		return nil
	}
	req := &domain.DeleteSinkEndpointUsageReq{
		SinkEndpointID: bizChangefeedEntity.BizPLEndpointID.Uint64(),
		BusinessType:   entity.BizPLEndpointUsageBusinessTypeChangefeed,
		BusinessID:     bizChangefeedEntity.ID.Uint64(),
	}
	return m.sinkEndpointModel.DeleteSinkEndpointUsage(ctx, req)
}

func (m *BizChangefeedModelImpl) preCheckDeleteBizChangefeed(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if err := m.checkBizChangefeedStatus(
		ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionDelete); err != nil {
		return err.Chain("check biz_changefeed status")
	}
	return nil
}

//nolint:dupl
func (m *BizChangefeedModelImpl) PauseBizChangefeed(
	ctx context.Context, req *domain.PauseBizChangefeedReq,
) (err bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	if err = m.preCheckPauseBizChangefeed(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("precheck pause biz_changefeed")
	}
	nextStatus, err := m.calBizChangefeedNextStatus(ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionPause)
	if err != nil {
		return err.Chain("calculate next status")
	}
	defer func() { m.handleEventHook(ctx, err, entity.PauseChangefeed, req.UserID, bizChangefeedEntity) }()

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		bizChangefeedEntity.OperationStatus = nextStatus
		if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		if err := m.pauseKernelChangefeed(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("pause kernel changefeed")
		}
		return nil
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}
	return nil
}

func (m *BizChangefeedModelImpl) pauseKernelChangefeed(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	if kernelChangefeed.KernelMetrics.FeedState == entity.ChangefeedKernelStatusStopped {
		return nil
	}
	_, err = m.kernelChangefeedModelFactory.GetModel().
		PauseChangefeed(ctx, &domain.PauseChangefeedReq{
			ChangefeedID: kernelChangefeed.ID.Uint64(),
		})
	if err != nil {
		return err.Chain("pause changefeed")
	}
	return nil
}

func (m *BizChangefeedModelImpl) preCheckPauseBizChangefeed(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if err := m.checkBizChangefeedStatus(
		ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionPause); err != nil {
		return err.Chain("check biz_changefeed status")
	}
	return nil
}

//nolint:dupl
func (m *BizChangefeedModelImpl) ResumeBizChangefeed(
	ctx context.Context, req *domain.ResumeBizChangefeedReq,
) (err bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	if err = m.preCheckResumeBizChangefeed(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("precheck resume biz_changefeed")
	}
	nextStatus, err := m.calBizChangefeedNextStatus(ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionResume)
	if err != nil {
		return err.Chain("calculate next status")
	}
	defer func() { m.handleEventHook(ctx, err, entity.ResumeChangefeed, req.UserID, bizChangefeedEntity) }()

	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		bizChangefeedEntity.OperationStatus = nextStatus
		if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		if err := m.resumeKernelChangefeed(ctx, bizChangefeedEntity); err != nil {
			return err.Chain("resume kernel changefeed")
		}
		return nil
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}
	return nil
}

func (m *BizChangefeedModelImpl) resumeKernelChangefeed(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	if kernelChangefeed.KernelMetrics.FeedState == entity.ChangefeedKernelStatusNormal {
		log.GetLogger(ctx).Warn("kernel changefeed status is normal, doesn't need to resume",
			zap.Uint64("biz_changefeed_id", bizChangefeedEntity.ID.Uint64()),
			zap.Uint64("kernel_changefeed_id", kernelChangefeed.ID.Uint64()))
		return nil
	}

	_, err = m.kernelChangefeedModelFactory.GetModel().
		ResumeChangefeed(ctx, &domain.ResumeChangefeedReq{
			ChangefeedID: kernelChangefeed.ID.Uint64(),
		})
	if err != nil {
		return err.Chain("resume changefeed")
	}
	return nil
}

func (m *BizChangefeedModelImpl) preCheckResumeBizChangefeed(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if err := m.checkBizChangefeedStatus(
		ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionResume); err != nil {
		return err.Chain("check biz_changefeed status")
	}
	return nil
}

func (m *BizChangefeedModelImpl) EditBizChangefeed(
	ctx context.Context, req *domain.EditBizChangefeedReq,
) (err bizerr.BizError) {
	logger := log.GetLogger(ctx)
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	if err = m.preCheckEditBizChangefeed(ctx, req, bizChangefeedEntity, kernelChangefeed); err != nil {
		return err.Chain("precheck edit biz_changefeed")
	}
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, bizChangefeedEntity.ClusterID.Uint64())
	if err != nil {
		return err.Chain("get cluster")
	}

	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(req.DownstreamType)
	if err != nil {
		return err.Chain("get downstream adaptor")
	}
	if err = downstreamAdaptor.CompleteEditParams(
		ctx, req.DownstreamParameter, bizChangefeedEntity.DownstreamParams); err != nil {
		return err.Chain("manage edit param")
	}
	if req.DryRun {
		logger.Info("dry run edit biz_changefeed successfully")
		return nil
	}

	defer func() { m.handleEventHook(ctx, err, entity.EditChangefeed, req.UserID, bizChangefeedEntity) }()
	transErr := m.transProvider.Transaction(ctx, func(ctx context.Context) error {
		if err := m.updateBizChangefeedEntityForEdit(ctx, req, bizChangefeedEntity); err != nil {
			return err.Chain("update biz_changefeed entity")
		}
		if err := m.editKernelChangefeed(
			ctx, cluster, downstreamAdaptor, bizChangefeedEntity, kernelChangefeed); err != nil {
			return err.Chain("edit kernel changefeed")
		}
		return nil
	})
	if transErr != nil {
		return convertTransactionError(transErr)
	}
	return nil
}

func (m *BizChangefeedModelImpl) preCheckEditBizChangefeed(
	ctx context.Context,
	req *domain.EditBizChangefeedReq,
	bizChangefeedEntity *entity.BizChangefeed,
	kernelChangefeed *entity.TiCDCChangefeed,
) bizerr.BizError {
	if kernelChangefeed.KernelMetrics.FeedState != entity.ChangefeedKernelStatusStopped &&
		kernelChangefeed.KernelMetrics.FeedState != entity.ChangefeedKernelStatusFailed {
		return bizerr.New(bizerr.ErrCodeParams, "kernel changefeed is not stopped, can't edit changefeed")
	}

	downstreamType, err := convertDownstreamType(req.DownstreamType)
	if err != nil {
		return err.Chain("convert downstream type")
	}
	if downstreamType != bizChangefeedEntity.DownstreamType {
		return bizerr.New(bizerr.ErrCodeParams, "can't edit downstream type")
	}

	if err = m.checkBizChangefeedStatus(
		ctx, bizChangefeedEntity.OperationStatus, bizChangefeedActionEdit); err != nil {
		return err.Chain("check biz_changefeed status")
	}
	return nil
}

func (m *BizChangefeedModelImpl) updateBizChangefeedEntityForEdit(
	ctx context.Context, req *domain.EditBizChangefeedReq, bizChangefeedEntity *entity.BizChangefeed,
) bizerr.BizError {
	if req.Name != "" {
		bizChangefeedEntity.Name = req.Name
	}
	downstreamParamsBytes, err := json.Marshal(req.DownstreamParameter)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeParams, "marshal downstream params")
	}
	bizChangefeedEntity.DownstreamParams = string(downstreamParamsBytes)
	if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("update biz_changefeed entity")
	}
	return nil
}

func (m *BizChangefeedModelImpl) EditBizChangefeedForOps(ctx context.Context, bizChangefeedID uint64) bizerr.BizError {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, bizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(
		domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeedEntity.DownstreamType))
	if err != nil {
		return err.Chain("get downstream adaptor")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, bizChangefeedEntity.ClusterID.Uint64())
	if err != nil {
		return err.Chain("get cluster")
	}
	if err := m.editKernelChangefeed(
		ctx, cluster, downstreamAdaptor, bizChangefeedEntity, kernelChangefeed); err != nil {
		return err.Chain("edit kernel changefeed")
	}
	return nil
}

func (m *BizChangefeedModelImpl) editKernelChangefeed(
	ctx context.Context,
	cluster *core.Cluster,
	downstreamAdaptor biz_changefeed.DownstreamAdaptor,
	bizChangefeedEntity *entity.BizChangefeed,
	kernelChangefeed *entity.TiCDCChangefeed,
) bizerr.BizError {
	kernelChangefeedConfig, err := downstreamAdaptor.AssembleChangefeedConfig(
		ctx, cluster, bizChangefeedEntity)
	if err != nil {
		return err.Chain("assemble kernel changefeed config")
	}
	_, err = m.kernelChangefeedModelFactory.GetModel().ModifyChangefeedConfig(
		ctx, &domain.ModifyChangefeedConfigReq{
			ChangefeedID: kernelChangefeed.ID.Uint64(),
			Config:       kernelChangefeedConfig,
		})
	if err != nil {
		return err.Chain("edit kernel changefeed")
	}
	return nil
}

func (m *BizChangefeedModelImpl) TestBizChangefeed(
	ctx context.Context, req *domain.TestBizChangefeedReq,
) bizerr.BizError {
	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(req.DownstreamType)
	if err != nil {
		return err.Chain("get downstream adaptor")
	}
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, req.ClusterID)
	if err != nil {
		return err.Chain("get cluster")
	}
	if err := downstreamAdaptor.TestConnection(ctx, cluster, req.DownstreamParameter); err != nil {
		return err.Chain("test connection")
	}
	return nil
}

func (m *BizChangefeedModelImpl) GetClusterBizChangefeedRCU(
	ctx context.Context, req *domain.GetClusterBizChangefeedRCUReq,
) (*domain.GetClusterBizChangefeedRCUResp, bizerr.BizError) {
	filterOp := genGetClusterBizChangefeedRCUFilterOp(req)
	bizChangefeedEntities, err := m.bizChangefeedRepo.List(ctx, filterOp)
	if err != nil {
		return nil, err.Chain("list biz_changefeeds")
	}
	if len(bizChangefeedEntities) == 0 {
		return &domain.GetClusterBizChangefeedRCUResp{}, nil
	}

	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	listSpecResp, err := m.machineManager.ListSpecifications(ctx, &domain.ListSpecificationsReq{
		InfraProviderType: &cluster.InfraProviderType,
	})
	if err != nil {
		return nil, err.Chain("list specifications")
	}
	specMap := specifications2Map(listSpecResp.Specifications)

	resp := &domain.GetClusterBizChangefeedRCUResp{}
	for i := range bizChangefeedEntities {
		if spec, ok := specMap[bizChangefeedEntities[i].SpecificationCode]; ok {
			resp.RCUNumber += spec.GetRCUNumber()
		} else {
			return nil, bizerr.New(
				bizerr.ErrCodeUnknown, "unknown specification code: %s",
				bizChangefeedEntities[i].SpecificationCode)
		}
	}
	return resp, nil
}

func (m *BizChangefeedModelImpl) CalBizChangefeedFee(
	ctx context.Context, req *domain.CalBizChangefeedFeeReq,
) (*domain.CalBizChangefeedFeeResp, bizerr.BizError) {
	specResp, err := m.machineManager.Query(ctx, &domain.QuerySpecificationReq{
		Code: req.SpecificationCode,
	})
	if err != nil {
		return nil, err.Chain("query specification code")
	}

	labels := map[string]string{
		config.BillingLabelTenantID:  strconv.FormatUint(req.TenantID, 10),
		config.BillingLabelProjectID: strconv.FormatUint(req.ProjectID, 10),
	}
	unit := fmt.Sprintf(config.TCUCostUnitPattern, req.CloudProvider, req.RegionName)
	costs := billing.UnitCostsInput{
		CostTimeStamp: time.Now().Unix(),
		Unit:          unit,
		Cost:          int(int64(specResp.Specification.GetRCUNumber()) * req.Duration),
		Labels:        labels,
	}
	fee, feeErr := m.billingHandler.CalFee(ctx, req.TenantID, &costs)
	if feeErr != nil {
		return nil, bizerr.WrapError(feeErr, "calculate changefeed fee")
	}
	return &domain.CalBizChangefeedFeeResp{
		Fee:    fee.Fee,
		RawFee: fee.RawFee,
	}, nil
}

func (m *BizChangefeedModelImpl) GetChangefeedDetail(
	ctx context.Context, req *domain.GetBizChangefeedDetailReq,
) (*domain.GetBizChangefeedDetailResp, bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz_changefeed")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return nil, err.Chain("query kernel changefeed")
	}
	resp, err := m.kernelChangefeedModelFactory.GetModel().
		GetChangefeedDetail(ctx, &domain.GetChangefeedDetailReq{ChangefeedID: kernelChangefeed.ID})
	if err != nil {
		return nil, err.Chain("query ticdc changefeed detail")
	}
	return &domain.GetBizChangefeedDetailResp{Changefeed: resp.Changefeed}, nil
}

func (m *BizChangefeedModelImpl) GetTiCDCCluster(
	ctx context.Context, req *domain.GetBizChangefeedClusterReq,
) (*domain.GetBizChangefeedClusterResp, bizerr.BizError) {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, req.BizChangefeedID)
	if err != nil {
		return nil, err.Chain("load biz_changefeed")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return nil, err.Chain("query kernel changefeed")
	}
	resp, err := m.kernelChangefeedModelFactory.GetModel().
		GetTiCDCCluster(ctx, &domain.GetChangefeedClusterReq{ChangefeedID: kernelChangefeed.ID})
	if err != nil {
		return nil, err.Chain("query ticdc cluster")
	}
	return &domain.GetBizChangefeedClusterResp{TiCDCCluster: resp.TiCDCCluster}, nil
}

func (m *BizChangefeedModelImpl) ListTiCDCClusters(
	ctx context.Context, req *domain.ListTiCDCClusterReq,
) (*domain.ListTiCDCClusterResp, bizerr.BizError) {
	return m.kernelChangefeedModelFactory.GetModel().
		ListTiCDCClusters(ctx, req)
}

func (m *BizChangefeedModelImpl) ListBizChangefeedSpecifications(
	ctx context.Context, req *domain.ListBizChangefeedSpecificationsReq,
) (*domain.ListBizChangefeedSpecificationsResp, bizerr.BizError) {
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	listSpecResp, err := m.machineManager.ListSpecifications(ctx, &domain.ListSpecificationsReq{
		InfraProviderType: &cluster.InfraProviderType,
	})
	if err != nil {
		return nil, err.Chain("list specifications")
	}
	specs := listSpecResp.Specifications

	resp := &domain.ListBizChangefeedSpecificationsResp{
		Specifications: make([]*domain.BizChangefeedSpecification, 0, len(specs)),
		Total:          uint32(len(specs)),
	}
	for i := range specs {
		resp.Specifications = append(resp.Specifications, specificationEntity2Domain(specs[i]))
	}
	sort.Slice(resp.Specifications, func(i, j int) bool {
		return resp.Specifications[i].RCUNumber < resp.Specifications[j].RCUNumber
	})
	return resp, nil
}

func (m *BizChangefeedModelImpl) BackfillChangefeedConfig(
	ctx context.Context, bizChangefeedID uint64,
) bizerr.BizError {
	bizChangefeedEntity, err := m.bizChangefeedRepo.Load(ctx, bizChangefeedID)
	if err != nil {
		return err.Chain("load biz_changefeed")
	}
	downstreamAdaptor, err := m.downstreamAdaptorFactory.Get(
		domain.TiCDCChangefeedDownStreamType2ConnectorType(bizChangefeedEntity.DownstreamType))
	if err != nil {
		return err.Chain("get downstream adaptor")
	}
	kernelChangefeed, err := m.queryKernelChangefeed(ctx, bizChangefeedEntity.ID.Uint64())
	if err != nil {
		return err.Chain("query kernel changefeed")
	}
	resp, err := m.kernelChangefeedModelFactory.GetModel().
		GetChangefeedDetail(ctx, &domain.GetChangefeedDetailReq{ChangefeedID: kernelChangefeed.ID})
	if err != nil {
		return err.Chain("get kernel changefeed config")
	}

	changefeedBytes, mErr := resp.Changefeed.MarshalJSON()
	if mErr != nil {
		return bizerr.WrapError(mErr, "marshal changefeed config")
	}
	regionalChangefeed := new(ticdc_v1.Changefeed)
	if err := protojson.Unmarshal(changefeedBytes, regionalChangefeed); err != nil {
		return bizerr.WrapError(err, "unmarshal changefeed config")
	}
	downstreamParams, changed, err := downstreamAdaptor.BackfillConfig(
		ctx, bizChangefeedEntity.DownstreamParams, regionalChangefeed.Config)
	if err != nil {
		return err.Chain("backfill changefeed config")
	}
	if !changed {
		return nil
	}

	bizChangefeedEntity.DownstreamParams = downstreamParams
	if err := m.bizChangefeedRepo.Update(ctx, bizChangefeedEntity); err != nil {
		return err.Chain("update biz_changefeed entity")
	}
	log.GetLogger(ctx).Info(
		"config changed and backfill changefeed config successfully",
		zap.Uint64("biz_changefeed_id", bizChangefeedID))
	return nil
}

func (m *BizChangefeedModelImpl) DescribeAdminProfile(
	ctx context.Context, req *domain.DescribeAdminProfileReq,
) (*domain.DescribeAdminProfileResp, bizerr.BizError) {
	cluster, err := m.coreCli.GetAvailableClusterByID(ctx, req.ClusterID)
	if err != nil {
		return nil, err.Chain("get cluster")
	}
	roleReq := &domain.GetTiCDCClusterRoleReq{
		TenantID:      req.TenantID,
		ProjectID:     req.ProjectID,
		TiDBClusterID: req.ClusterID,
		Region:        cluster.RegionName,
		CloudProvider: domain.CloudProviderType(cluster.CloudProviderName),
	}
	roleInfo, err := m.cdcClusterModel.GetRole(ctx, roleReq)
	if err != nil {
		return nil, err.Chain("get ticdc cluster role")
	}

	resp := &domain.DescribeAdminProfileResp{
		InfraProviderType: domain.InfraProviderType(cluster.InfraProviderType),
	}
	switch cluster.CloudProviderName {
	case string(domain.CloudProviderTypeAWS):
		resp.AWSAccountID = roleInfo.CloudAccountID
		resp.AWSExternalID = roleInfo.AWSExternalID
	case string(domain.CloudProviderTypeGCP):
		resp.GCPServiceAccount = roleInfo.CloudIAMIdentity
	}
	return resp, nil
}

func (m *BizChangefeedModelImpl) DescribeSchemaTable(ctx context.Context,
	req *domain.DescribeSchemaTableReq,
) (*domain.DescribeSchemaTableResp, bizerr.BizError) {
	return m.regionalKernelChangefeedModel.DescribeSchemaTable(ctx, req)
}

func (c *BizChangefeedModelImpl) ListChangefeedInstanceTypes(
	ctx context.Context, req *dataflow.ChangefeedInstanceTypeListReq,
) (*dataflow.ChangefeedInstanceTypes, bizerr.BizError) {
	instanceTypes, err := c.changefeedInstanceTypeRepo.ListChangefeedInstanceType(ctx, repo.ChangefeedInstanceTypeFilter{
		BusinessType: entity.ChangefeedInstanceTypeBusinessTypeNormal,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "failed to get changefeed instance type")
	}
	var items []*dataflow.ChangefeedInstanceType
	for _, t := range instanceTypes {
		items = append(items,
			&dataflow.ChangefeedInstanceType{
				Name:     t.Name,
				RcuNum:   t.RCUNum,
				Replicas: t.Replicas,
				RpsLimit: t.RPSLimit,
			})
	}
	return &dataflow.ChangefeedInstanceTypes{
		Total: uint32(len(items)),
		Items: items,
	}, nil
}

func (m *BizChangefeedModelImpl) queryKernelChangefeed(
	ctx context.Context, bizChangefeedID uint64,
) (*entity.TiCDCChangefeed, bizerr.BizError) {
	bizType := domain.TiCDCClusterBizTypeChangefeed
	// regionalKernelChangefeedModel can get all the changefeeds
	resp, err := m.regionalKernelChangefeedModel.QueryChangefeed(ctx, &domain.QueryChangefeedReq{
		Filter: &domain.ListChangefeedFilter{
			BizType:     &bizType,
			BusinessIDs: []uint64{bizChangefeedID},
		},
	})
	if err != nil {
		return nil, err
	}
	return &resp.ChangefeedEntity, nil
}

func (m *BizChangefeedModelImpl) listKernelChangefeeds(
	ctx context.Context, bizChangefeedIDs []uint64,
) ([]*entity.TiCDCChangefeed, bizerr.BizError) {
	bizType := domain.TiCDCClusterBizTypeChangefeed
	// regionalKernelChangefeedModel can get all the changefeeds
	resp, err := m.regionalKernelChangefeedModel.ListChangefeeds(ctx, &domain.ListChangefeedsReq{
		Filter: &domain.ListChangefeedFilter{
			BizType:     &bizType,
			BusinessIDs: bizChangefeedIDs,
		},
		Pager: domain.NewPager(domain.PageSizeOp(int64(len(bizChangefeedIDs)))),
	})
	if err != nil {
		return nil, err
	}

	kernelChangefeedEntities := make([]*entity.TiCDCChangefeed, 0, len(resp.Items))
	for i := range resp.Items {
		kernelChangefeedEntities = append(kernelChangefeedEntities, &resp.Items[i].ChangefeedEntity)
	}
	return kernelChangefeedEntities, nil
}

func (m *BizChangefeedModelImpl) checkBizChangefeedStatus(
	ctx context.Context, bizChangefeedStatus entity.BizChangefeedOperationStatus, action string,
) bizerr.BizError {
	currentStatus := status_matrix.CurrentStatus{
		Status: string(bizChangefeedStatus),
	}
	if !m.statusMatrix.IsActionFriendly2Status(currentStatus, action) {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"current status %s, not allowed to %s", bizChangefeedStatus, action)
	}
	return nil
}

func (m *BizChangefeedModelImpl) createBizChangefeedWorkflow(
	ctx context.Context, bizChangefeedEntity *entity.BizChangefeed, dagName, dagActionName string,
) (string, bizerr.BizError) {
	vars, tags := buildBizChangefeedDagVarsTags(bizChangefeedEntity.ID.Uint64(), dagActionName)
	dagIns, err := m.workflowCli.RunDagWithTags(dagName, vars, tags)
	if err != nil {
		return "", bizerr.New(bizerr.ErrCodeUnknown, "run dag error: %s", err.Error())
	}
	return dagIns.ID, nil
}

func (m *BizChangefeedModelImpl) calBizChangefeedNextStatus(
	ctx context.Context, bizChangefeedStatus entity.BizChangefeedOperationStatus, action string,
) (entity.BizChangefeedOperationStatus, bizerr.BizError) {
	currentStatus := status_matrix.CurrentStatus{
		Status: string(bizChangefeedStatus),
	}
	nextStatus := m.statusMatrix.CalculateNextStatusForAction(currentStatus, action)
	if nextStatus == nil {
		return "", bizerr.New(
			bizerr.ErrCodeUnknown,
			"calculate next status failed, current status: %s, action: %s", bizChangefeedStatus, action)
	}
	return entity.BizChangefeedOperationStatus(nextStatus.Status), nil
}

func (m *BizChangefeedModelImpl) checkSpecificationCode(ctx context.Context, specificationCode string) bizerr.BizError {
	_, err := m.machineManager.Query(ctx, &domain.QuerySpecificationReq{
		Code: specificationCode,
	})
	return err
}

func (m *BizChangefeedModelImpl) handleEventHook(
	ctx context.Context,
	err error,
	event entity.DataflowEventType,
	userID uint64,
	bizChangefeedEntity *entity.BizChangefeed,
) {
	if err != nil {
		return
	}
	m.hooker.HandleBizChangefeedEvent(ctx, event, userID, bizChangefeedEntity)
}

func buildBizChangefeedDagVarsTags(bizChangefeedID uint64, dagAction string) (vars, tags map[string]string) {
	vars = map[string]string{
		workflow.KeyBizChangefeedID: strconv.FormatUint(bizChangefeedID, 10),
	}
	tags = map[string]string{
		workflow.BusinessID:        strconv.FormatUint(bizChangefeedID, 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeBizChangefeed),
		workflow.BusinessActionKey: dagAction,
	}
	return
}

func convertDownstreamType(
	connectorType dataflow.ConnectorType,
) (entity.TiCDCChangefeedDownStreamType, bizerr.BizError) {
	downstreamType := domain.ConnectorType2TiCDCChangefeedDownStreamType(connectorType)
	if downstreamType == "" {
		return "", bizerr.New(bizerr.ErrCodeParams, "unsupported downstream type %s", connectorType.String())
	}
	return downstreamType, nil
}

func convertTransactionError(err error) bizerr.BizError {
	if err == nil {
		return nil
	}

	var bizErr bizerr.BizError
	if errors.As(err, &bizErr) {
		return bizErr
	}
	return bizerr.New(bizerr.ErrCodeDB, "execute transaction error: %s", err.Error())
}

func bizChangefeedEntity2Domain(bizChangefeedEntity *entity.BizChangefeed) *domain.BizChangefeed {
	return &domain.BizChangefeed{
		BizChangefeedID:     bizChangefeedEntity.ID.Uint64(),
		Name:                bizChangefeedEntity.Name,
		CreatedAt:           bizChangefeedEntity.CreatedAt,
		SpecificationCode:   bizChangefeedEntity.SpecificationCode,
		DownstreamType:      bizChangefeedEntity.DownstreamType,
		DownstreamParameter: bizChangefeedEntity.DownstreamParams,
		NetworkType:         bizChangefeedEntity.NetworkType,
		NetworkInfo:         bizChangefeedEntity.NetworkInfo,
		DeployType:          bizChangefeedEntity.DeployType,
		OperationStatus:     bizChangefeedEntity.OperationStatus,
		ErrorMessage:        bizChangefeedEntity.ErrMsg,
		ClusterID:           bizChangefeedEntity.ClusterID.Uint64(),
		ProjectID:           bizChangefeedEntity.ProjectID.Uint64(),
		TenantID:            bizChangefeedEntity.TenantID.Uint64(),
		TiCDCClusterID:      bizChangefeedEntity.TiCDCClusterID.Uint64(),
	}
}

func fillKernelChangefeedInfo(bizChangefeed *domain.BizChangefeed, kernelChangefeedEntity *entity.TiCDCChangefeed) {
	bizChangefeed.ChangefeedStatus = kernelChangefeedEntity.KernelMetrics.FeedState
	bizChangefeed.CheckpointTSO = kernelChangefeedEntity.KernelMetrics.CheckpointTSO
	if err := kernelChangefeedEntity.KernelMetrics.RunningError; err != nil {
		bizChangefeed.ErrorMessage = err.Error()
	}
}

func genListBizChangefeedsFilterOp(req *domain.ListBizChangefeedsReq) repo.BizChangefeedFilterOp {
	return func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
		qs = qs.ClusterIDEq(entity.ID(req.ClusterID)).
			OperationStatusNotIn(
				entity.BizChangefeedOperationStatusDeleting, entity.BizChangefeedOperationStatusDeleted)
		if req.DownstreamType != nil {
			qs = qs.DownstreamTypeEq(*req.DownstreamType)
		}
		return qs
	}
}

func genListBizChangefeedsPagerOp(req *domain.ListBizChangefeedsReq) repo.BizChangefeedFilterOp {
	pager := domain.NewPager(domain.PageNumberOp(int64(req.Page)), domain.PageSizeOp(int64(req.PerPage)))
	offset, limit := pager.CalculateOffsetAndLimit()
	return func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
		return qs.Offset(int(offset)).Limit(int(limit)).OrderDescByCreatedAt()
	}
}

func genGetClusterBizChangefeedRCUFilterOp(req *domain.GetClusterBizChangefeedRCUReq) repo.BizChangefeedFilterOp {
	return func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
		return qs.ClusterIDEq(entity.ID(req.ClusterID)).
			OperationStatusNotIn(
				entity.BizChangefeedOperationStatusCreating, entity.BizChangefeedOperationStatusDeleted)
	}
}

func kernelChangefeedEntities2Map(
	kernelChangefeedEntities []*entity.TiCDCChangefeed,
) map[uint64]*entity.TiCDCChangefeed {
	kernelChangefeedMap := make(map[uint64]*entity.TiCDCChangefeed, len(kernelChangefeedEntities))
	for i := range kernelChangefeedEntities {
		kernelChangefeedMap[kernelChangefeedEntities[i].BusinessID.Uint64()] = kernelChangefeedEntities[i]
	}
	return kernelChangefeedMap
}

// TODO(ideascf): use domain.BizChangefeedSpecification directly to avoid this conversion
func specificationEntity2Domain(spec *domain.TiCDCClusterSpecification) *domain.BizChangefeedSpecification {
	return &domain.BizChangefeedSpecification{
		SpecificationCode: spec.Code,
		Replicas:          spec.Replicas,
		RPSLimit:          spec.RPSLimit,
		RCUNumber:         spec.GetRCUNumber(),
	}
}

func specifications2Map(specs []*domain.TiCDCClusterSpecification) map[string]*domain.TiCDCClusterSpecification {
	specMap := make(map[string]*domain.TiCDCClusterSpecification, len(specs))
	for i := range specs {
		specMap[specs[i].Code] = specs[i]
	}
	return specMap
}

func getNetworkInfoStrAndType(
	networkInfo *dataflow.NetworkInfo) (string,
	entity.ChangefeedDownstreamNetworkType, bizerr.BizError,
) {
	if networkInfo == nil {
		return "", entity.ChangefeedDownstreamNetworkTypeInternet, nil
	}
	infoBytes, err := json.Marshal(networkInfo)
	if err != nil {
		return "", entity.ChangefeedDownstreamNetworkTypeUnknown,
			bizerr.New(bizerr.ErrCodeParams, "marshal network info error: %s", err.Error())
	}
	var networkType entity.ChangefeedDownstreamNetworkType
	switch networkInfo.NetworkType {
	case dataflow.NetworkType_NetworkType_Public:
		networkType = entity.ChangefeedDownstreamNetworkTypeInternet
	case dataflow.NetworkType_NetworkType_VPC_Peering:
		networkType = entity.ChangefeedDownstreamNetworkTypeVPCPeering
	case dataflow.NetworkType_NetworkType_Private_Link:
		networkType = entity.ChangefeedDownstreamNetworkTypePrivateLink
	case dataflow.NetworkType_NetworkType_Private_Service_Connect:
		networkType = entity.ChangefeedDownstreamNetworkTypePrivateServiceConnect
	case dataflow.NetworkType_NetworkType_Manually_Private_Link:
		networkType = entity.DownstreamNetworkTypeManuallyPrivateLink
	case dataflow.NetworkType_NetworkType_Manually_Private_Service_Connect:
		networkType = entity.DownstreamNetworkTypeManuallyPrivateServiceConnect
	default:
		return "", entity.ChangefeedDownstreamNetworkTypeUnknown, bizerr.New(bizerr.ErrCodeParams, "invalid network type")
	}
	return string(infoBytes), networkType, nil
}