package cluster

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"sync"

	workflowentity "github.com/tidbcloud/fastflow/pkg/entity"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/metrics"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/pkg/utils"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

func (svc *clusterServiceImpl) List(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceListRequest,
) (results []*globalserverlessv1.ServerlessServiceCluster, err error) {
	ctx, span := Tracer.Start(ctx, "Domain.ListCluster")
	defer span.End()

	defer func() {
		metrics.RequestListClusterCount.Inc()
		metrics.RequestClusterResultCount.Add(float64(len(results)))
		if len(results) > 100 {
			metrics.LargeClusterQueryResultRequestCount.Inc()
			tilog.Warn(ctx, "dangerous cluster query, too many results hit", zap.Int("length", len(results)))
		}
	}()

	// Load all regions and clusters from the repository.
	filteredClusters, err := svc.loadClustersAndRegionsFromRepo(ctx, req)
	if err != nil {
		return nil, err
	}

	if filteredClusters.empty() {
		return []*globalserverlessv1.ServerlessServiceCluster{}, nil
	}

	singleRegionQueryFunc := func(region *entity.Region, _ *sync.WaitGroup, resChan chan<- *adapter.RegionalResult[*regionalserverlessv1.ServerlessRegionalServiceClusterInfo]) {
		regionalRes := &adapter.RegionalResult[*regionalserverlessv1.ServerlessRegionalServiceClusterInfo]{}
		newReq, ok := proto.Clone(req).(*globalserverlessv1.ServerlessServiceListRequest)
		if !ok {
			regionalRes.Err = bizerr.New(bizerr.ErrCodeUnknown, "clone request failed")
			resChan <- regionalRes
			return
		}
		if newReq.Filter != nil {
			// set only the cluster id that belongs to the target region.
			if cs, ok := filteredClusters.region2Cluster[region.Id]; ok && cs != nil {
				newReq.Filter.ClusterIds = cs
				ps := make([]uint64, 0, len(cs))
				for _, c := range cs {
					ps = append(ps, filteredClusters.activeCluster[c].ProjectId)
				}
				newReq.Filter.ProjectIds = ps
			}
		}
		if region.State != globalcommonv1.Region_ACTIVE {
			regionalRes.Err = bizerr.New(bizerr.ErrCodeUnknown, "the region is unavailable")
			resChan <- regionalRes
			return
		}

		if isPGOnlyQuery(newReq, filteredClusters.activeCluster) {
			regionalRes.Items = []*regionalserverlessv1.ServerlessRegionalServiceClusterInfo{}
			resChan <- regionalRes
			return
		}

		clusters, err := svc.regionalService.ListCluster(ctx, region, newReq)
		if err != nil {
			regionalRes.Err = bizerr.WrapError(err, "list cluster failed")
			resChan <- regionalRes
			return
		}
		regionalRes.Items = clusters
		resChan <- regionalRes
	}

	resChan := adapter.QueryMultiRegional(filteredClusters.regions, singleRegionQueryFunc)
	resultClusterIDMap := make(map[string]struct{})

	// Compose the result.
	for result := range resChan {
		if result.Err != nil {
			return nil, result.Err
		}

		for _, cluster := range result.Items {
			if _, ok := filteredClusters.activeCluster[cluster.ClusterId]; !ok {
				tilog.Warn(ctx, "the cluster is not in global service", zap.String("cluster", cluster.ClusterId))
				continue
			}

			globalCluster, err := svc.clusterRegionalToGlobal(
				ctx,
				cluster,
				filteredClusters.activeCluster[cluster.ClusterId],
				req.GetFilter().GetShowDetails(),
			)
			if err != nil {
				tilog.Warn(
					ctx,
					"convert regional cluster to global cluster failed",
					zap.String("cluster", cluster.ClusterId),
					zap.Error(err),
				)
				return nil, err
			}

			if globalCluster != nil {
				// found the cluster in the global service
				// delete the cluster from the creating cluster map
				delete(filteredClusters.creatingCluster, cluster.ClusterId)
				resultClusterIDMap[cluster.ClusterId] = struct{}{}
				results = append(results, globalCluster)
			}
		}
	}

	for clusterID, cluster := range filteredClusters.activeCluster {
		if !isPGStarterCluster(cluster) {
			continue
		}
		if cluster.GetUserStatus() == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
			continue
		}
		if _, ok := resultClusterIDMap[clusterID]; ok {
			continue
		}
		globalCluster := cvt.ClusterEnToGlobal(cluster)
		svc.enrichPGEndpointFromSys9(ctx, cluster, globalCluster)
		results = append(results, globalCluster)
	}

	// full creating cluster
	// not found in regional service, but found in global service
	// add more fields with workflow instance
	for _, cluster := range filteredClusters.creatingCluster {
		cls, err := svc.mergeCreatingCluster(ctx, cluster, req.GetFilter().GetShowDetails())
		if err != nil {
			tilog.Warn(ctx, "merge creating cluster failed", zap.String("cluster", cluster.ClusterId), zap.Error(err))
			return nil, err
		}
		if cls != nil {
			results = append(results, cls)
		}
	}

	results = append(results, filteredClusters.deletedCluster...)

	return results, nil
}

func (svc *clusterServiceImpl) Count(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceCountClustersRequest,
) (int64, error) {
	if req.Filter == nil {
		return svc.clusterRepo.Count(ctx, &repository.ClusterQueryParameter{})
	}

	return svc.clusterRepo.Count(ctx, &repository.ClusterQueryParameter{
		ClusterIds: req.Filter.ClusterIds,
		TenantIds:  req.Filter.TenantIds,
		ProjectIds: req.Filter.ProjectIds,
		Deleted:    shouldShowDeletedClusters(req.Filter),
	})
}

func (svc *clusterServiceImpl) clusterRegionalToGlobal(
	ctx context.Context,
	regionalCluster *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	repoCluster *entity.Cluster,
	showDetails bool,
) (*globalserverlessv1.ServerlessServiceCluster, error) {
	// convert regional cluster to global cluster
	globalCluster := cvt.ClusterRegionalToGlobal(regionalCluster)
	if repoCluster != nil {
		globalCluster.ProjectId = repoCluster.ProjectId
		globalCluster.HasSetPwd = repoCluster.HasSetPwd
		if regionalCluster.State == regionalserverlessv1.ClusterState_CLUSTER_STATE_READY &&
			(repoCluster.CreationFinishedAt == nil || repoCluster.CreationFinishedAt.IsZero()) {
			globalCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING
		}

		if globalCluster.UserStatus == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING &&
			globalCluster.GetCluster().GetRestoreCluster() {
			tilog.Info(ctx, "cluster is restoring", zap.String("cluster", globalCluster.ClusterId))
			globalCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_RECOVERING
		}

		if repoCluster.CreationFinishedAt != nil {
			globalCluster.CreationFinishedAt = proto.Int64(repoCluster.CreationFinishedAt.Unix())
		}
		if repoCluster.Region != nil &&
			repoCluster.Region.DisplayName != "" &&
			globalCluster.GetCluster().GetRegion() != nil {
			globalCluster.Cluster.Region.DisplayName = repoCluster.Region.DisplayName
		}
	}

	// get update workflow instance
	workflowInstances, err := svc.flow.ListRunningDagInstances(
		ctx,
		[]string{common.UpdateCluster},
		repoCluster.GetClusterId(),
	)
	if err != nil {
		tilog.Error(
			ctx,
			"get update workflow instance failed",
			zap.String("cluster", repoCluster.GetClusterId()),
			zap.Error(err),
		)
	}

	for _, ins := range workflowInstances {
		switch ins.DagID {
		case common.UpdateCluster:
			tilog.Info(
				ctx,
				"get update workflow instance",
				zap.String("cluster", repoCluster.GetClusterId()),
				zap.Any("instance", ins),
			)
			if updateReq := unmarshal(ctx, repoCluster.GetClusterId(), ins, &globalserverlessv1.ServerlessServiceUpdateRequest{}); updateReq != nil {
				if pe := updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetCluster().GetEndpoints().GetPublic(); pe != nil {
					if globalCluster.Cluster.Endpoints == nil {
						globalCluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
					}
					if globalCluster.Cluster.Endpoints.Public == nil {
						globalCluster.Cluster.Endpoints.Public = &regionalserverlessv1.PublicEndpoint{}
					}
					globalCluster.Cluster.Endpoints.Public.Disabled = pe.GetDisabled()
					globalCluster.Cluster.Endpoints.Public.AuthorizedNetwork = pe.GetAuthorizedNetwork()
				}
				if pe := updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetCluster().GetEndpoints().GetPrivate(); pe != nil {
					if globalCluster.Cluster.Endpoints == nil {
						globalCluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
					}
					if globalCluster.Cluster.Endpoints.Private == nil {
						globalCluster.Cluster.Endpoints.Private = &regionalserverlessv1.PrivateEndpoint{}
					}
					globalCluster.Cluster.Endpoints.Private.Disabled = pe.GetDisabled()
					globalCluster.Cluster.Endpoints.Private.AuthorizedNetwork = pe.GetAuthorizedNetwork()
				}
			}
		default:
			tilog.Warn(
				ctx,
				"unknown workflow instance",
				zap.String("cluster", repoCluster.GetClusterId()),
				zap.String("instance", ins.DagID),
			)
		}
	}

	if showDetails {
		if err := svc.appendClusterDetailField(ctx, globalCluster, workflowInstances); err != nil {
			tilog.Warn(
				ctx,
				"get cluster detail field failed",
				zap.String("cluster", regionalCluster.ClusterId),
				zap.Error(err),
			)
			return globalCluster, err
		}
	}

	if isPGStarterCluster(repoCluster) {
		svc.enrichPGEndpointFromSys9(ctx, repoCluster, globalCluster)
	}

	return globalCluster, nil
}

func (svc *clusterServiceImpl) mergeCreatingCluster(
	ctx context.Context,
	globalCluster *globalserverlessv1.ServerlessServiceCluster,
	showDetails bool,
) (*globalserverlessv1.ServerlessServiceCluster, error) {
	if globalCluster == nil ||
		globalCluster.GetUserStatus() != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
		tilog.Warn(ctx, "cluster is not creating", zap.String("cluster", globalCluster.GetClusterId()))
		return globalCluster, nil
	}

	// get creating workflow instance
	workflowInstances, err := svc.flow.ListRunningDagInstances(
		ctx,
		[]string{common.CreateCluster},
		globalCluster.ClusterId,
	)
	if err != nil {
		return globalCluster, bizerr.WrapGormError(err, "get create workflow instance failed")
	}

	for _, ins := range workflowInstances {
		switch ins.DagID {
		case common.CreateCluster:
			tilog.Info(
				ctx,
				"get create workflow instance",
				zap.String("cluster", globalCluster.ClusterId),
				zap.Any("instance", ins),
			)
			if createReq := unmarshal(ctx, globalCluster.ClusterId, ins, &globalserverlessv1.ServerlessServiceCreateRequest{}); createReq != nil {
				globalCluster.Cluster.State = regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING
				globalCluster.Cluster.ZoneRedundantEnabled = createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetZoneRedundantEnabled()
				globalCluster.Cluster.EnhancedEncryptionEnabled = createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetEncryptionConfig().
					GetEnhancedEncryptionEnabled()
				globalCluster.Cluster.AutoScaling = createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetAutoScaling()
				if endpoints := createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetEndpoints(); endpoints != nil {
					if globalCluster.Cluster.Endpoints == nil {
						globalCluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
					}
					if public := endpoints.GetPublic(); public != nil {
						if globalCluster.Cluster.Endpoints.Public == nil {
							globalCluster.Cluster.Endpoints.Public = &regionalserverlessv1.PublicEndpoint{}
						}
						globalCluster.Cluster.Endpoints.Public.Disabled = public.GetDisabled()
						globalCluster.Cluster.Endpoints.Public.AuthorizedNetwork = public.GetAuthorizedNetwork()
					}
					if private := endpoints.GetPrivate(); private != nil {
						if globalCluster.Cluster.Endpoints.Private == nil {
							globalCluster.Cluster.Endpoints.Private = &regionalserverlessv1.PrivateEndpoint{}
						}
						globalCluster.Cluster.Endpoints.Private.Disabled = private.GetDisabled()
						globalCluster.Cluster.Endpoints.Private.AuthorizedNetwork = private.GetAuthorizedNetwork()
					}
				}
			}
		default:
			tilog.Warn(
				ctx,
				"unknown workflow instance",
				zap.String("cluster", globalCluster.ClusterId),
				zap.String("instance", ins.DagID),
			)
		}
	}

	if showDetails {
		if err := svc.appendClusterDetailField(ctx, globalCluster, workflowInstances); err != nil {
			return globalCluster, err
		}
	}

	return globalCluster, nil
}

type listClustersResult struct {
	regions         []*entity.Region
	activeCluster   map[string]*entity.Cluster
	creatingCluster map[string]*globalserverlessv1.ServerlessServiceCluster
	deletedCluster  []*globalserverlessv1.ServerlessServiceCluster
	region2Cluster  map[globalcommonv1.Region_Name][]string
}

func (r *listClustersResult) empty() bool {
	return len(r.regions) == 0 && len(r.activeCluster) == 0 && len(r.deletedCluster) == 0 && len(r.creatingCluster) == 0
}

func (svc *clusterServiceImpl) loadClustersAndRegionsFromRepo(
	ctx context.Context,
	req *globalserverlessv1.ServerlessServiceListRequest,
) (*listClustersResult, error) {
	ctx, span := Tracer.Start(ctx, "Domain.loadClustersAndRegionsFromRepo")
	defer span.End()

	if req.Filter == nil {
		metrics.ClusterQueryWithoutFilterCount.Inc()
		return nil, errors.New("request without filter")
	}

	parameter := repository.ClusterQueryParameter{
		ClusterIds:         req.Filter.ClusterIds,
		EngineTypes:        req.Filter.EngineTypes,
		TenantIds:          req.Filter.TenantIds,
		CloudProviderNames: req.Filter.CloudProviderNames,
		ProjectIds:         req.Filter.ProjectIds,
		Deleted:            shouldShowDeletedClusters(req.Filter),
	}

	if !utils.IsZero(reflect.ValueOf(parameter)) {
		clusters, err := svc.clusterRepo.Query(ctx, &parameter)
		if err != nil {
			return nil, err
		}

		return svc.filterRegionAndClusters(clusters), nil
	}

	metrics.ClusterQueryWithoutFilterCount.Inc()
	tilog.Warn(ctx, "request without filter")

	return svc.loadAllRegions(ctx)
}

func shouldShowDeletedClusters(filter *globalserverlessv1.ClusterFilter) bool {
	if filter == nil {
		return true
	}

	for _, state := range filter.States {
		if state == regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETING ||
			state == regionalserverlessv1.ClusterState_CLUSTER_STATE_DELETED {
			return true
		}
	}

	if filter.Deleted != nil {
		return *filter.Deleted
	}

	return true
}

func isPGStarterCluster(cluster *entity.Cluster) bool {
	if cluster == nil {
		return false
	}
	if cluster.ServicePlan != globalcommonv1.ServicePlan_Starter {
		return false
	}
	return strings.EqualFold(cluster.EngineType, "POSTGRESQL_COMPATIBLE")
}

func isPGOnlyQuery(req *globalserverlessv1.ServerlessServiceListRequest, clusters map[string]*entity.Cluster) bool {
	if req == nil || req.GetFilter() == nil {
		return false
	}
	clusterIDs := req.GetFilter().GetClusterIds()
	if len(clusterIDs) == 0 {
		return false
	}
	for _, clusterID := range clusterIDs {
		cluster, ok := clusters[clusterID]
		if !ok || !isPGStarterCluster(cluster) {
			return false
		}
	}
	return true
}

func (svc *clusterServiceImpl) enrichPGEndpointFromSys9(
	ctx context.Context,
	cluster *entity.Cluster,
	globalCluster *globalserverlessv1.ServerlessServiceCluster,
) {
	if svc == nil || cluster == nil || globalCluster == nil || globalCluster.GetCluster() == nil {
		return
	}

	if globalCluster.GetCluster().GetEndpoints().GetPublic().GetHost() != "" {
		return
	}

	if cluster.ExternalInstanceID == "" {
		tilog.Warn(ctx, "skip enrich pg endpoint from sys9: external instance id is empty", zap.String("cluster", cluster.GetClusterId()))
		return
	}

	connectionInfo, err := svc.sys9Lifecycle.GetManagedDatabaseConnectionInfo(ctx, cluster, cluster.ExternalInstanceID)
	if err != nil {
		tilog.Warn(
			ctx,
			"enrich pg endpoint from sys9 failed",
			zap.String("cluster", cluster.GetClusterId()),
			zap.String("service_instance_id", cluster.ExternalInstanceID),
			zap.Error(err),
		)
		return
	}
	if connectionInfo == nil || connectionInfo.Host == "" || connectionInfo.Port <= 0 {
		tilog.Warn(
			ctx,
			"enrich pg endpoint from sys9 got empty endpoint",
			zap.String("cluster", cluster.GetClusterId()),
			zap.String("service_instance_id", cluster.ExternalInstanceID),
		)
		return
	}

	if globalCluster.Cluster.Endpoints == nil {
		globalCluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
	}
	if globalCluster.Cluster.Endpoints.Public == nil {
		globalCluster.Cluster.Endpoints.Public = &regionalserverlessv1.PublicEndpoint{}
	}
	globalCluster.Cluster.Endpoints.Public.Host = connectionInfo.Host
	globalCluster.Cluster.Endpoints.Public.Port = int64(connectionInfo.Port)
}

func (svc *clusterServiceImpl) filterRegionAndClusters(clusters []*entity.Cluster) *listClustersResult {
	out := &listClustersResult{
		deletedCluster:  make([]*globalserverlessv1.ServerlessServiceCluster, 0),
		creatingCluster: map[string]*globalserverlessv1.ServerlessServiceCluster{},
		regions:         make([]*entity.Region, 0),
		activeCluster:   map[string]*entity.Cluster{},
		region2Cluster:  map[globalcommonv1.Region_Name][]string{},
	}

	regionMap := make(map[globalcommonv1.Region_Name]struct{})
	for _, c := range clusters {
		if !c.DeletedAt.IsZero() {
			out.deletedCluster = append(out.deletedCluster, cvt.DeletedClusterEnToGlobal(c))
		} else {
			if c.UserStatus == globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
				out.creatingCluster[c.ClusterId] = cvt.ClusterEnToGlobal(c)
			}
			out.activeCluster[c.ClusterId] = c
			_, ok := regionMap[c.Region.Id]
			if !ok {
				regionMap[c.Region.Id] = struct{}{}
				out.regions = append(out.regions, c.Region)
			}

			cs, ok := out.region2Cluster[c.Region.Id]
			if !ok {
				cs = make([]string, 0)
			}
			cs = append(cs, c.ClusterId)
			out.region2Cluster[c.Region.Id] = cs
		}
	}

	return out
}

func (svc *clusterServiceImpl) loadAllRegions(ctx context.Context) (*listClustersResult, error) {
	regions, err := svc.regionRepo.LoadRegions(ctx, nil)
	if err != nil {
		return nil, err
	}

	return &listClustersResult{
		regions: regions,
	}, nil
}

func (svc *clusterServiceImpl) appendClusterDetailField(
	ctx context.Context,
	globalCluster *globalserverlessv1.ServerlessServiceCluster,
	workflowInstances []*workflowentity.DagInstance,
) error {
	/* rcu > 0 essential plan, needless to get spend limit */
	if globalCluster.GetCluster().GetAutoScaling().GetMaxRcu() == 0 {
		spendLimit, err := svc.serverlessFacade.GetSpendLimit(
			ctx,
			globalCluster.ClusterId,
			globalCluster.TenantId,
		)
		if err != nil {
			if globalCluster.GetUserStatus() != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
				return err
			}
			tilog.Warn(
				ctx,
				"skip creating cluster get spend limit failed",
				zap.String("cluster", globalCluster.ClusterId),
				zap.Error(err),
			)
		} else {
			// get spend limit success
			globalCluster.SpendLimit = spendLimit
		}
	}

	policy, err := svc.brFacade.GetAutoBackupPolicy(
		ctx,
		globalCluster.ClusterId,
		globalCluster.TenantId,
		globalCluster.ProjectId,
	)
	if err != nil {
		if globalCluster.GetUserStatus() != globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING {
			return err
		}
		tilog.Warn(
			ctx,
			"skip creating cluster get backup policy",
			zap.String("cluster", globalCluster.ClusterId),
			zap.Error(err),
		)
	} else {
		globalCluster.BackupPolicy = policy
	}

	for _, ins := range workflowInstances {
		switch ins.DagID {
		case common.UpdateCluster:
			tilog.Info(
				ctx,
				"get update workflow instance",
				zap.String("cluster", globalCluster.ClusterId),
				zap.Any("instance", ins),
			)
			if updateReq := unmarshal(ctx, globalCluster.ClusterId, ins, &globalserverlessv1.ServerlessServiceUpdateRequest{}); updateReq != nil {
				if updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetSpendLimit().
					GetMonthly() !=
					0 {
					if globalCluster.SpendLimit == nil {
						globalCluster.SpendLimit = &globalserverlessv1.SpendLimit{}
					}
					globalCluster.SpendLimit = updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetSpendLimit()
				}

				if updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetBackupPolicy() != nil {
					globalCluster.BackupPolicy = updateReq.(*globalserverlessv1.ServerlessServiceUpdateRequest).Cluster.GetBackupPolicy()
				}
			}
		case common.CreateCluster:
			tilog.Info(
				ctx,
				"get create workflow instance",
				zap.String("cluster", globalCluster.ClusterId),
				zap.Any("instance", ins),
			)
			if createReq := unmarshal(ctx, globalCluster.ClusterId, ins, &globalserverlessv1.ServerlessServiceCreateRequest{}); createReq != nil {
				if createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetSpendLimit() != nil {
					globalCluster.SpendLimit = createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetSpendLimit()
				}

				if createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetBackupPolicy() != nil {
					globalCluster.BackupPolicy = createReq.(*globalserverlessv1.ServerlessServiceCreateRequest).GetBackupPolicy()
				}
			}
		default:
			tilog.Warn(
				ctx,
				"unknown workflow instance",
				zap.String("cluster", globalCluster.ClusterId),
				zap.String("instance", ins.DagID),
			)
		}
	}

	return nil
}

func unmarshal(ctx context.Context, clusterId string, ins *workflowentity.DagInstance, r proto.Message) proto.Message {
	req, ok := ins.Vars[common.KeyRequestBody]
	if !ok {
		tilog.Warn(ctx, "request body not found", zap.String("cluster", clusterId))
		return nil
	}

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

	return r
}