package cluster

import (
	"context"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/samber/lo"
	workflowentity "github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.einride.tech/aip/pagination"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	bffclusterv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/bff/cluster_management/v1beta1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	openapiserverlessv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/serverless/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/app/bff/orchestrator"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func (s *Server) ListClusters(
	ctx context.Context,
	request *bffclusterv1beta1.ListClustersRequest,
) (*bffclusterv1beta1.ListClustersResponse, error) {
	// Extract request meta.
	_, err := metadata.ExtractRequestMetaWithCheckPermission(ctx)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "failed to extract request metadata")
	}

	// check if the request is valid
	if request.ProjectId == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "project_id is required")
	}

	projectId, err := strconv.ParseUint(request.ProjectId, 10, 64)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "project_id is invalid")
	}

	// page and sort cluster from mgmt repo
	pageToken, err := pagination.ParsePageToken(request)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeParams, "invalid page token")
	}

	pageSize := request.GetPageSize()
	if pageSize <= 0 {
		pageSize = common.ListDefaultPageSize
	}

	clusterFilter := &repository.ClusterFilter{
		ProjectIds: []uint64{projectId},
		Name:       request.GetName(),
	}

	switch request.GetClusterType() {
	case bffclusterv1beta1.Cluster_SERVERLESS:
		clusterFilter.DeployType = entity.DeployTypeServerless
	case bffclusterv1beta1.Cluster_DEDICATED:
		clusterFilter.DeployType = entity.DeployTypeDedicated
	}

	// count
	count, err := s.clusterRepo.CountClusterByFilter(ctx, clusterFilter)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeDB, "failed to count clusters")
	}

	if pageToken.Offset >= count || count == 0 {
		return &bffclusterv1beta1.ListClustersResponse{
			Clusters:      nil,
			TotalSize:     int32(count),
			NextPageToken: "",
		}, nil
	}

	allClusters, err := s.clusterRepo.QueryClusterByFilterAndPage(
		ctx,
		clusterFilter,
		int(pageToken.Offset),
		int(pageSize),
	)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, bizerr.ErrCodeDB, "failed to query clusters")
	}

	serverlessIds := lo.Map(lo.Filter(allClusters, func(item *entity.Cluster, _ int) bool {
		return item.DeployType == entity.DeployTypeServerless
	}), func(item *entity.Cluster, _ int) uint64 {
		return item.ClusterId
	})

	dedicatedIds := lo.Map(lo.Filter(allClusters, func(item *entity.Cluster, _ int) bool {
		return item.DeployType == entity.DeployTypeDedicated
	}), func(item *entity.Cluster, _ int) uint64 {
		return item.ClusterId
	})

	// list serverless clusters
	var serverlessResp *openapiserverlessv1beta1.ListClustersResponse
	if len(serverlessIds) > 0 {
		serverlessResp, err = s.openAPIServerlessClusterServer.ListClusters(
			ctx,
			&openapiserverlessv1beta1.ListClustersRequest{
				Filter: fmt.Sprintf("projectId = %s AND clusterId = %q",
					request.GetProjectId(),
					strings.Join(lo.Map(serverlessIds, func(id uint64, _ int) string {
						return strconv.FormatUint(id, 10)
					}), ","),
				),
				PageSize: pageSize,
			},
		)

		if err != nil {
			return nil, err
		}
	}

	// list dedicated clusters
	var dedicatedResp *openapidedicatedv1beta1.ListClustersResponse
	if len(dedicatedIds) > 0 {
		dedicatedResp, err = s.openAPIDedicatedClusterServer.ListClusters(
			ctx,
			&openapidedicatedv1beta1.ListClustersRequest{
				ProjectId: request.GetProjectId(),
				ClusterIds: lo.Map(dedicatedIds, func(id uint64, _ int) string {
					return strconv.FormatUint(id, 10)
				}),
				PageSize: pageSize,
			},
		)
		if err != nil {
			return nil, err
		}
	}

	clusters := make([]*bffclusterv1beta1.Cluster, 0, len(allClusters))

	// append all serverless clusters
	for _, cluster := range serverlessResp.GetClusters() {
		clusters = append(clusters, &bffclusterv1beta1.Cluster{
			ClusterType: bffclusterv1beta1.Cluster_SERVERLESS,
			Cluster:     &bffclusterv1beta1.Cluster_ServerlessCluster{ServerlessCluster: cluster},
		})
	}

	// append all dedicated clusters
	for _, cluster := range dedicatedResp.GetClusters() {
		// Caution(@yifan): filter insufficient clusters.
		//   Since insufficient filter only exists in BFF and not in open api, the filter is hard coded here.
		if request.GetOnlyResourceInsufficient() {
			if cluster.GetAnnotations()[common.InsufficientVmResourcesKey] != "true" {
				continue
			}
		}

		clusters = append(clusters, &bffclusterv1beta1.Cluster{
			ClusterType: bffclusterv1beta1.Cluster_DEDICATED,
			Cluster:     &bffclusterv1beta1.Cluster_DedicatedCluster{DedicatedCluster: cluster},
		})
	}

	// sort by create time desc
	slices.SortFunc(clusters, func(a, b *bffclusterv1beta1.Cluster) int {
		ca, cb := getCreateAt(a), getCreateAt(b)
		if ca.After(cb) {
			return -1
		}
		if ca.Before(cb) {
			return 1
		}
		return 0
	})

	// merge in workflow state
	for _, cluster := range clusters {
		if cluster.ClusterType == bffclusterv1beta1.Cluster_DEDICATED {
			dagIns, err := s.orchestrator.ListRunningDagInstancesByNamePrefix(
				ctx,
				fmt.Sprintf(
					"%s_%s_%s",
					workflow.DynamicDagPrefix,
					orchestrator.DagConsoleAPIScale,
					cluster.GetDedicatedCluster().GetClusterId(),
				),
				cluster.GetDedicatedCluster().ClusterId,
			)
			// if list running dag instances failed, log the error and continue
			if err != nil {
				tilog.Error(ctx, "list running dag instances failed", zap.Error(err))
				continue
			}
			cluster.Cluster = &bffclusterv1beta1.Cluster_DedicatedCluster{
				DedicatedCluster: s.mergeInWorkflowState(ctx, cluster.GetDedicatedCluster(), dagIns),
			}
		}
	}

	resp := &bffclusterv1beta1.ListClustersResponse{
		Clusters:  clusters,
		TotalSize: int32(count),
	}

	if pageToken.Offset+int64(pageSize) < count {
		pageToken.Offset += int64(pageSize)
		resp.NextPageToken = pageToken.String()
	}

	return resp, nil
}

func (s *Server) mergeInWorkflowState(
	ctx context.Context,
	cluster *openapidedicatedv1beta1.Cluster,
	runningDagInstances []*workflowentity.DagInstance,
) *openapidedicatedv1beta1.Cluster {
	if len(runningDagInstances) == 0 {
		return cluster
	}

	ins := runningDagInstances[0]

	if strings.HasPrefix(
		ins.DagID,
		fmt.Sprintf("%s_%s_%s", workflow.DynamicDagPrefix, orchestrator.DagConsoleAPIScale, cluster.GetClusterId()),
	) {
		out := unmarshal(ctx, cluster.ClusterId, ins, &bffclusterv1beta1.AdvancedModifyDedicatedClusterRequest{})
		if out == nil {
			return cluster
		}

		advancedCluster := out.(*bffclusterv1beta1.AdvancedModifyDedicatedClusterRequest).Cluster

		tilog.Warn(ctx, "mergeInWorkflowState", zap.Any("advancedCluster", advancedCluster))
		cluster = s.mergeAdvancedCluster(ctx, cluster, advancedCluster)
	}

	return cluster
}

func (s *Server) mergeAdvancedCluster(
	ctx context.Context,
	cluster *openapidedicatedv1beta1.Cluster,
	advancedCluster *bffclusterv1beta1.AdvancedModifyDedicatedClusterRequest_Cluster,
) *openapidedicatedv1beta1.Cluster {
	// check cluster id
	if cluster.GetClusterId() != advancedCluster.ClusterId {
		tilog.Warn(
			ctx,
			"cluster id not match",
			zap.String("cluster_id", cluster.GetClusterId()),
			zap.String("advanced_cluster_id", advancedCluster.ClusterId),
		)
		return cluster
	}

	// set cluster status
	cluster.State = openapicommonv1beta1.ClusterState_MODIFYING

	// need delete tidb node group ids
	// key: tidb node group id
	deleteTidbNodeGroupIds := make(
		map[string]struct{},
		len(advancedCluster.GetTidbNodeSetting().GetDeleteTidbNodeGroupIds()),
	)
	for _, id := range advancedCluster.GetTidbNodeSetting().GetDeleteTidbNodeGroupIds() {
		deleteTidbNodeGroupIds[id] = struct{}{}
	}

	// need update tidb node group ids
	// key: tidb node group id
	// value: new node count
	updateTidbNodeGroupIds := make(
		map[string]int32,
		len(advancedCluster.GetTidbNodeSetting().GetUpdateTidbNodeGroups()),
	)
	for _, g := range advancedCluster.GetTidbNodeSetting().GetUpdateTidbNodeGroups() {
		updateTidbNodeGroupIds[g.GetTidbNodeGroupId()] = g.GetNodeCount()
	}

	// update create tidb node groups
	// key: tidb node group display name
	// value: count
	createTidbNodeGroups := make(map[string]int32, len(advancedCluster.GetTidbNodeSetting().GetCreateTidbNodeGroups()))
	for _, g := range advancedCluster.GetTidbNodeSetting().GetCreateTidbNodeGroups() {
		createTidbNodeGroups[g.GetDisplayName()] = g.GetNodeCount()
	}

	// updated tiproxy setting groups
	// key: tidb node group id
	// value: tiproxy setting
	updateTiProxyGroups := make(
		map[string]*openapidedicatedv1beta1.TidbNodeGroup_TiProxySetting,
		len(advancedCluster.GetTidbNodeSetting().GetUpdateTidbNodeGroups()),
	)
	for _, g := range advancedCluster.GetTidbNodeSetting().GetUpdateTidbNodeGroups() {
		if g.GetTiproxySetting() != nil {
			updateTiProxyGroups[g.GetTidbNodeGroupId()] = g.GetTiproxySetting()
		}
	}

	// created tiproxy setting groups
	// key: tidb node group display name
	// value: tiproxy setting
	createTiProxyGroups := make(
		map[string]*openapidedicatedv1beta1.TidbNodeGroup_TiProxySetting,
		len(advancedCluster.GetTidbNodeSetting().GetCreateTidbNodeGroups()),
	)
	for _, g := range advancedCluster.GetTidbNodeSetting().GetCreateTidbNodeGroups() {
		if g.GetTiproxySetting() != nil {
			createTiProxyGroups[g.GetDisplayName()] = g.GetTiproxySetting()
		}
	}

	// merge tidb node spec
	if advancedCluster.GetTidbNodeSetting().GetNodeSpecKey() != "" &&
		cluster.GetTidbNodeSetting().GetNodeSpecKey() != advancedCluster.GetTidbNodeSetting().GetNodeSpecKey() {
		// set new node spec key
		if cluster.TidbNodeSetting == nil {
			cluster.TidbNodeSetting = &openapidedicatedv1beta1.Cluster_TidbNodeSetting{}
		}

		cluster.TidbNodeSetting.NodeSpecKey = advancedCluster.GetTidbNodeSetting().GetNodeSpecKey()
		// get node spec display name
		nodeSpec, err := s.openAPIDedicatedRegionServer.GetNodeSpec(ctx, &openapidedicatedv1beta1.GetNodeSpecRequest{
			NodeSpecKey:   advancedCluster.GetTidbNodeSetting().GetNodeSpecKey(),
			ProjectId:     cluster.GetLabels()[common.ProjectIDLabelKey],
			ComponentType: openapidedicatedv1beta1.ComponentType_TIDB,
			RegionId:      cluster.GetRegionId(),
		})
		if err != nil {
			tilog.Error(ctx, "get node spec failed", zap.Error(err))
		} else {
			cluster.TidbNodeSetting.NodeSpecDisplayName = nodeSpec.GetDisplayName()
		}
	}

	// full all tidb node groups node spec key
	// already exists tidb node groups use new node spec key and set remaining deletion node count
	tidbGroups := make(
		[]*openapidedicatedv1beta1.TidbNodeGroup,
		0,
		len(cluster.GetTidbNodeSetting().GetTidbNodeGroups()),
	)
	for _, g := range cluster.GetTidbNodeSetting().GetTidbNodeGroups() {
		// filter out deleted tidb node groups
		if _, ok := deleteTidbNodeGroupIds[g.GetTidbNodeGroupId()]; ok {
			continue
		}

		// update tidb node group
		if newCount, ok := updateTidbNodeGroupIds[g.GetTidbNodeGroupId()]; ok {
			if g.GetNodeCount() != newCount {
				g.NodeCount = newCount
				g.State = openapidedicatedv1beta1.TidbNodeGroup_MODIFYING
			}
		}

		// filter already exists tidb node groups from create tidb node groups
		delete(createTidbNodeGroups, g.GetDisplayName())

		// check if node spec key changed
		if cluster.GetTidbNodeSetting().GetNodeSpecKey() != g.GetNodeSpecKey() {
			g.NodeSpecKey = advancedCluster.GetTidbNodeSetting().GetNodeSpecKey()
			g.NodeSpecDisplayName = cluster.GetTidbNodeSetting().GetNodeSpecDisplayName()
			g.NodeChangingProgress = &openapidedicatedv1beta1.Cluster_NodeChangingProgress{
				MatchingNodeSpecNodeCount:  0,
				RemainingDeletionNodeCount: g.GetNodeCount(),
			}
			g.State = openapidedicatedv1beta1.TidbNodeGroup_MODIFYING
		}

		if tiproxySetting, ok := updateTiProxyGroups[g.GetTidbNodeGroupId()]; ok {
			if g.GetTiproxySetting().GetNodeSpecKey() != tiproxySetting.GetNodeSpecKey() ||
				g.GetTiproxySetting().GetNodeCount() != tiproxySetting.GetNodeCount() {
				g.TiproxySetting = tiproxySetting
				g.State = openapidedicatedv1beta1.TidbNodeGroup_MODIFYING
			}
			delete(createTiProxyGroups, g.GetDisplayName())
		}
		tidbGroups = append(tidbGroups, g)
	}

	// add new tidb node groups
	for name, count := range createTidbNodeGroups {
		newTidbGroup := &openapidedicatedv1beta1.TidbNodeGroup{
			Name:            "", // not created yet
			TidbNodeGroupId: "", // not created yet
			ClusterId:       cluster.GetClusterId(),
			DisplayName:     name,
			NodeCount:       count,
			NodeChangingProgress: &openapidedicatedv1beta1.Cluster_NodeChangingProgress{
				MatchingNodeSpecNodeCount:  0,
				RemainingDeletionNodeCount: 0,
			},
			NodeSpecKey:         cluster.GetTidbNodeSetting().GetNodeSpecKey(),
			NodeSpecDisplayName: cluster.GetTidbNodeSetting().GetNodeSpecDisplayName(),
			State:               openapidedicatedv1beta1.TidbNodeGroup_MODIFYING,
		}
		if tiproxySetting, ok := createTiProxyGroups[name]; ok {
			newTidbGroup.TiproxySetting = tiproxySetting
		}
		tidbGroups = append(tidbGroups, newTidbGroup)
	}

	// replace tidb node groups
	cluster.TidbNodeSetting.TidbNodeGroups = tidbGroups

	// merge tikv node setting
	if advancedCluster.TikvNodeSetting != nil {
		if cluster.TikvNodeSetting == nil {
			cluster.TikvNodeSetting = &openapidedicatedv1beta1.Cluster_StorageNodeSetting{}
		}

		changed := false
		matchingCount := cluster.GetTikvNodeSetting().GetNodeCount()
		deleteingCount := int32(0)

		// check if node count changed
		if advancedCluster.GetTikvNodeSetting().GetNodeCount() != 0 &&
			advancedCluster.GetTikvNodeSetting().GetNodeCount() != cluster.GetTikvNodeSetting().GetNodeCount() {
			changed = true
			if advancedCluster.GetTikvNodeSetting().GetNodeCount() < cluster.GetTikvNodeSetting().GetNodeCount() {
				matchingCount = advancedCluster.GetTikvNodeSetting().GetNodeCount()
				deleteingCount = cluster.GetTikvNodeSetting().
					GetNodeCount() -
					advancedCluster.GetTikvNodeSetting().
						GetNodeCount()
			}

			cluster.TikvNodeSetting.NodeCount = advancedCluster.GetTikvNodeSetting().GetNodeCount()
		}

		// check if storage size changed
		if advancedCluster.GetTikvNodeSetting().GetStorageSizeGi() != 0 &&
			advancedCluster.GetTikvNodeSetting().GetStorageSizeGi() != cluster.GetTikvNodeSetting().GetStorageSizeGi() {
			changed = true

			// if storage size changed, reset matching count
			matchingCount = 0

			cluster.TikvNodeSetting.StorageSizeGi = advancedCluster.GetTikvNodeSetting().GetStorageSizeGi()
		}

		// check if data store iops changed
		if advancedCluster.GetTikvNodeSetting().DataStoreIops != nil &&
			advancedCluster.GetTikvNodeSetting().GetDataStoreIops() != cluster.GetTikvNodeSetting().GetDataStoreIops() {
			changed = true

			cluster.TikvNodeSetting.DataStoreIops = lo.ToPtr(advancedCluster.GetTikvNodeSetting().GetDataStoreIops())
		}

		// check if raft store iops changed
		if advancedCluster.GetTikvNodeSetting().RaftStoreIops != nil &&
			advancedCluster.GetTikvNodeSetting().GetRaftStoreIops() != cluster.GetTikvNodeSetting().GetRaftStoreIops() {
			changed = true

			cluster.TikvNodeSetting.RaftStoreIops = lo.ToPtr(advancedCluster.GetTikvNodeSetting().GetRaftStoreIops())
		}

		if advancedCluster.GetTikvNodeSetting().GetNodeSpecKey() != "" &&
			cluster.GetTikvNodeSetting().GetNodeSpecKey() != advancedCluster.GetTikvNodeSetting().GetNodeSpecKey() {
			changed = true

			// if node spec key changed, reset matching count and deleteing count
			matchingCount = 0
			deleteingCount = 0

			// set new node spec key
			cluster.TikvNodeSetting.NodeSpecKey = advancedCluster.GetTikvNodeSetting().GetNodeSpecKey()
			// get node spec display name
			nodeSpec, err := s.openAPIDedicatedRegionServer.GetNodeSpec(
				ctx,
				&openapidedicatedv1beta1.GetNodeSpecRequest{
					NodeSpecKey:   advancedCluster.GetTikvNodeSetting().GetNodeSpecKey(),
					ProjectId:     cluster.GetLabels()[common.ProjectIDLabelKey],
					ComponentType: openapidedicatedv1beta1.ComponentType_TIKV,
					RegionId:      cluster.GetRegionId(),
				},
			)
			if err != nil {
				tilog.Error(ctx, "get node spec failed", zap.Error(err))
			} else {
				cluster.TikvNodeSetting.NodeSpecDisplayName = nodeSpec.GetDisplayName()
			}
		}

		if changed {
			cluster.TikvNodeSetting.NodeChangingProgress = &openapidedicatedv1beta1.Cluster_NodeChangingProgress{
				MatchingNodeSpecNodeCount:  matchingCount,
				RemainingDeletionNodeCount: deleteingCount,
			}
		}
	}

	// merge tiflash node setting
	if advancedCluster.TiflashNodeSetting != nil {
		if cluster.TiflashNodeSetting == nil {
			cluster.TiflashNodeSetting = &openapidedicatedv1beta1.Cluster_StorageNodeSetting{}
		}

		// set default storage type
		// todo support modify with storage type
		if cluster.GetTiflashNodeSetting().
			GetStorageType() ==
			openapidedicatedv1beta1.Cluster_StorageNodeSetting_STORAGE_TYPE_UNSPECIFIED {
			cluster.TiflashNodeSetting.StorageType = openapidedicatedv1beta1.Cluster_StorageNodeSetting_Standard
		}

		// delete tiflash node setting
		if advancedCluster.GetTiflashNodeSetting().NodeCount != nil &&
			advancedCluster.GetTiflashNodeSetting().GetNodeCount() == 0 {
			cluster.TiflashNodeSetting = nil
			return cluster
		}

		changed := false
		matchingCount := cluster.GetTiflashNodeSetting().GetNodeCount()
		deleteingCount := int32(0)

		// tiFlash node allow to change to 0
		// if set count is not nil, means user want to change the count
		if advancedCluster.GetTiflashNodeSetting().NodeCount != nil &&
			advancedCluster.GetTiflashNodeSetting().GetNodeCount() != cluster.GetTiflashNodeSetting().GetNodeCount() {
			changed = true
			if advancedCluster.GetTiflashNodeSetting().GetNodeCount() < cluster.GetTiflashNodeSetting().GetNodeCount() {
				matchingCount = advancedCluster.GetTiflashNodeSetting().GetNodeCount()
				deleteingCount = cluster.GetTiflashNodeSetting().
					GetNodeCount() -
					advancedCluster.GetTiflashNodeSetting().
						GetNodeCount()
			}
			cluster.TiflashNodeSetting.NodeCount = advancedCluster.GetTiflashNodeSetting().GetNodeCount()
		}

		if advancedCluster.GetTiflashNodeSetting().GetStorageSizeGi() != 0 &&
			advancedCluster.GetTiflashNodeSetting().
				GetStorageSizeGi() !=
				cluster.GetTiflashNodeSetting().
					GetStorageSizeGi() {
			changed = true

			// if storage size changed, reset matching
			matchingCount = 0

			cluster.TiflashNodeSetting.StorageSizeGi = advancedCluster.GetTiflashNodeSetting().GetStorageSizeGi()
		}

		if advancedCluster.GetTiflashNodeSetting().GetNodeSpecKey() != "" &&
			cluster.GetTiflashNodeSetting().
				GetNodeSpecKey() !=
				advancedCluster.GetTiflashNodeSetting().
					GetNodeSpecKey() {
			changed = true

			// if node spec key changed, reset matching count and deleteing count
			matchingCount = 0
			deleteingCount = 0

			// set new node spec key
			cluster.TiflashNodeSetting.NodeSpecKey = advancedCluster.GetTiflashNodeSetting().GetNodeSpecKey()
			// get node spec display name
			nodeSpec, err := s.openAPIDedicatedRegionServer.GetNodeSpec(
				ctx,
				&openapidedicatedv1beta1.GetNodeSpecRequest{
					NodeSpecKey:   advancedCluster.GetTiflashNodeSetting().GetNodeSpecKey(),
					ProjectId:     cluster.GetLabels()[common.ProjectIDLabelKey],
					ComponentType: openapidedicatedv1beta1.ComponentType_TIFLASH,
					RegionId:      cluster.GetRegionId(),
				},
			)
			if err != nil {
				tilog.Error(ctx, "get node spec failed", zap.Error(err))
			} else {
				cluster.TiflashNodeSetting.NodeSpecDisplayName = nodeSpec.GetDisplayName()
			}
		}

		if changed {
			cluster.TiflashNodeSetting.NodeChangingProgress = &openapidedicatedv1beta1.Cluster_NodeChangingProgress{
				MatchingNodeSpecNodeCount:  matchingCount,
				RemainingDeletionNodeCount: deleteingCount,
			}
		}
	}

	return cluster
}

func unmarshal(ctx context.Context, clusterId string, ins *workflowentity.DagInstance, r proto.Message) proto.Message {
	req, ok := ins.Vars[orchestrator.DagKeyRequest]
	if !ok {
		return nil
	}

	if err := protojson.Unmarshal([]byte(req.Value), r); err != nil {
		tilog.Error(
			ctx,
			"unmarshal request failed",
			zap.String("cluster", clusterId),
			zap.String("value", req.Value),
			zap.Error(err),
		)
		return nil
	}

	return r
}

func getCreateAt(c *bffclusterv1beta1.Cluster) time.Time {
	switch c.GetClusterType() {
	case bffclusterv1beta1.Cluster_SERVERLESS:
		return c.GetServerlessCluster().GetCreateTime().AsTime()
	case bffclusterv1beta1.Cluster_DEDICATED:
		return c.GetDedicatedCluster().GetCreateTime().AsTime()
	}

	return time.Now()
}