package cluster_svc

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/lithammer/shortuuid/v3"
	"github.com/shopspring/decimal"
	"github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.opentelemetry.io/otel"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

var Tracer = otel.Tracer("cluster")

const (
	exitReasonFailToGetClusterById = "failed to get cluster by id"
)

type SyncTidbPoolService struct {
	billingFacade       facade.BillingFacade
	mgmtFacade          facade.MgmtSvcFacade
	clusterRepo         repo.ClusterRepo
	clusterTidbPoolRepo repo.ClusterTidbPoolRepo
	accountFacade       facade.AccountFacade
	eventbusService     *eventbus.EventbusService
	canaryRepo          repo.CanaryRepo
	dbProvider          db.TransProvider
	branchRepo          repo.BranchRepo
	infraFacade         facade.InfraFacade
}

func NewSyncTidbPoolService(
	billingFacade facade.BillingFacade,
	mgmtFacade facade.MgmtSvcFacade,
	clusterRepo repo.ClusterRepo,
	clusterTidbPoolRepo repo.ClusterTidbPoolRepo,
	canaryRepo repo.CanaryRepo,
	accountFacade facade.AccountFacade,
	eventbusService *eventbus.EventbusService,
	dbProvider db.TransProvider,
	branchRepo repo.BranchRepo,
	infraFacade facade.InfraFacade,
) *SyncTidbPoolService {
	return &SyncTidbPoolService{
		billingFacade:       billingFacade,
		mgmtFacade:          mgmtFacade,
		clusterRepo:         clusterRepo,
		clusterTidbPoolRepo: clusterTidbPoolRepo,
		canaryRepo:          canaryRepo,
		accountFacade:       accountFacade,
		eventbusService:     eventbusService,
		dbProvider:          dbProvider,
		branchRepo:          branchRepo,
		infraFacade:         infraFacade,
	}
}

func (s *SyncTidbPoolService) Execute(ctx context.Context) (err error) {
	status, exitReason := metric.StatusOK, "-"
	ctx, span := Tracer.Start(ctx, "synctidbpool.Execute")
	defer func() {
		tracing.EndSpan(span, err)
		metric.TidbScheduleEventCounter.WithLabelValues("-", "sync-all", "-", "-", "-", status, exitReason).Inc()
	}()

	span.AddEvent("load unsynced records")
	records, err := s.clusterTidbPoolRepo.LoadUnsyncedRecords(ctx)
	if err != nil {
		exitReason = "failed to load unsync records from db"
		status = metric.StatusErr
		return err
	}
	for _, record := range records {
		/*
			if e := s.syncOneCluster(ctx, record); e != nil {
				err = multierr.Append(err, e)
			}*/
		if e := s.syncOneClusterV2(ctx, record); e != nil {
			err = multierr.Append(err, e)
		}
	}
	return err
}

func (s *SyncTidbPoolService) handleSpendLimitEvent(
	ctx context.Context,
	record *entity.ClusterTidbPool,
	cluster *tidbmgmtservicev1.Cluster,
	policy *tidbPoolPolicy,
	bizData *facade.ServerlessSpendLimitBizData,
) (_ *tidbPoolPolicy, exitReason string, err error) {
	if record.SpendLimitStatus == constant.SpendLimitEventStatusHandy {
		clusterEntity := &entity.Cluster{
			ID:                   cluster.ClusterId,
			KeyspaceID:           uint64(cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()),
			TenantID:             cluster.OrganizationId,
			ProjectID:            cluster.ProjectId,
			CloudProviderName:    cluster.CloudProvider,
			RegionName:           cluster.Region,
			CreatedTime:          time.Unix(cluster.GetServerlessCluster().Cluster.CreateTs, 0),
			CreationFinishedTime: nil,
			SpendLimit:           bizData.InvoiceLimit,
			FreeCredit:           bizData.FreeCredits,
			IsPaid:               bizData.InvoiceLimit.GreaterThan(decimal.Zero),
			CanaryMark:           cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoCanary] == constant.CanaryPoolMarkedFlag,
			RuMode:               getClusterRuModeFromPolicy(policy),
		}
		if cluster.GetServerlessCluster().CreationFinishedAt != nil {
			t := time.Unix(*cluster.GetServerlessCluster().CreationFinishedAt, 0)
			clusterEntity.CreationFinishedTime = &t
		}

		// upsert cluster with attribute `is_paid` after setting/modifying Spending Limit
		err = s.clusterRepo.Save(ctx, clusterEntity)
		if err != nil {
			exitReason = "failed to save cluster entity"
			return nil, exitReason, err
		}
		policy.SpendLimitStatus = constant.SpendLimitEventStatusHandy
	} else {
		err = s.clusterRepo.UpdateRUMode(ctx, cluster.ClusterId, getClusterRuModeFromPolicy(policy))
		if err != nil {
			exitReason = "failed to update cluster ru mode to repo"
			return nil, exitReason, err
		}
	}

	// if the cluster is marked as canary pool, generate canary pool policy
	c, err := s.clusterRepo.GetClusterByID(ctx, cluster.ClusterId)
	if err != nil {
		exitReason = exitReasonFailToGetClusterById
		return nil, exitReason, err
	}

	if c.CanaryMark {
		policy.PoolName = constant.IntoCanaryPool(policy.PoolName)
		policy.CanaryMark = constant.CanaryPoolMarkedFlag
	}

	if record.SpendLimitStatus != constant.SpendLimitEventStatusHandy &&
		record.SpendLimitStatus != policy.SpendLimitStatus && !utils.IsNotifyBlockTenantsAndEnv(cluster.OrganizationId) {
		/* send feishu alert */
		title := fmt.Sprintf(
			"[%s] Warnings, inconsistent spend_limit event state",
			strings.ToUpper(config.Global().ServerConfig.Stage),
		)
		tenant, _ := s.accountFacade.GetTenant(ctx, cluster.OrganizationId)
		if tenant == nil {
			tenant = &facade.Tenant{}
		}
		message := fmt.Sprintf("Tenant: %d / %s / %s / %v\n"+
			"Cluster: %d/%s / %s/%s\n"+
			"EventState: %s\n"+
			"BillingState=%s\n"+
			"SpendLimit=%s\n"+
			"FreeCredit=%s\n",
			cluster.OrganizationId, tenant.Name, tenant.Email, tenant.Kind,
			cluster.ClusterId, cluster.ClusterName, cluster.CloudProvider, cluster.Region,
			record.SpendLimitStatus,
			policy.SpendLimitStatus,
			bizData.InvoiceLimit.String(),
			bizData.FreeCredits.String())
		go func() {
			utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
		}()
	}

	return policy, "", nil
}

func (s *SyncTidbPoolService) syncOneClusterV2(ctx context.Context, record *entity.ClusterTidbPool) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx).With(zap.Uint64("cluster_id", record.ClusterID),
		zap.String("event_type", record.Type), zap.String("event_status", record.SpendLimitStatus))
	cluster, berr := s.mgmtFacade.GetClusterBypassCache(
		ctx,
		record.ClusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC,
	)
	if berr != nil {
		if berr.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			if e := s.clusterTidbPoolRepo.RemoveAllEvents(ctx, record.ClusterID); e != nil {
				berr = e
			}
		}
		logger.Error("failed to get cluster from mgmt-svc", zap.Error(berr))
		return berr
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		logger.Info("cluster is not in normal status, skip tidb pool sync")
		return nil
	}

	switch record.Type {
	case constant.TypeCanary:
		return s.handleCanaryEvents(ctx, logger, record, cluster)
	case constant.TypeSpendLimit:
		if slices.Contains(constant.SpendLimitValidStatuses, record.SpendLimitStatus) {
			// if essential plan cluster, only update cluster repo
			if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
				return s.handleEssentialPlanCluster(ctx, logger, record, cluster)
			}
			return s.handleStarterPlanCluster(ctx, logger, record, cluster)
		}
	}
	return s.clusterTidbPoolRepo.MarkAsSynced(ctx, record)
}

func (s *SyncTidbPoolService) handleCanaryEvents(
	ctx context.Context,
	logger *zap.Logger,
	record *entity.ClusterTidbPool,
	cluster *tidbmgmtservicev1.Cluster,
) error {
	logger.Info("handle canary events")
	originPolicy := &tidbPoolPolicy{
		PoolName:   cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoTidbPoolSelector],
		CanaryMark: cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoCanary],
	}
	newPolicy := &tidbPoolPolicy{}
	if record.CanaryMark {
		newPolicy.PoolName = constant.IntoCanaryPool(originPolicy.PoolName)
		newPolicy.CanaryMark = constant.CanaryPoolMarkedFlag
	} else {
		newPolicy.PoolName = constant.OutOfCanaryPool(originPolicy.PoolName)
		newPolicy.CanaryMark = constant.CanaryPoolUnmarkedFlag
	}
	updateCr := make(map[string]string)
	if originPolicy.PoolName != newPolicy.PoolName {
		updateCr[constant.AnnoTidbPoolSelector] = newPolicy.PoolName
	}
	if originPolicy.CanaryMark != newPolicy.CanaryMark {
		updateCr[constant.AnnoCanary] = newPolicy.CanaryMark
	}
	if len(updateCr) > 0 {
		if _, err := s.mgmtFacade.UpdateServerlessClusterAnnotations(ctx, cluster.ClusterId, updateCr); err != nil {
			return err
		}
		/* notify lark group for new schedule policy */
		if originPolicy.PoolName != newPolicy.PoolName &&
			constant.OutOfCanaryPool(newPolicy.PoolName) != constant.TidbPoolNameFree &&
			!utils.IsNotifyBlockTenantsAndEnv(cluster.OrganizationId) {
			title := fmt.Sprintf(
				"[%s] Update Tidb Schedule Policy",
				strings.ToUpper(config.Global().ServerConfig.Stage),
			)
			tenant, _ := s.accountFacade.GetTenant(ctx, cluster.OrganizationId)
			if tenant == nil {
				tenant = &facade.Tenant{}
			}
			message := fmt.Sprintf("Tenant: %d / %s / %s / %v\n"+
				"Cluster: %d/%s / %s/%s\n"+
				"EventType: %s\n"+
				"OldPool: %s\n"+
				"NewPool: %s\n",
				cluster.OrganizationId, tenant.Name, tenant.Email, tenant.Kind,
				cluster.ClusterId, cluster.ClusterName, cluster.CloudProvider, cluster.Region,
				record.Type,
				originPolicy.PoolName,
				newPolicy.PoolName,
			)
			go func() {
				utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
			}()
		}
		/* sync branches */
		{
			status := metric.StatusOK
			// can't solely rely on EventTypeResourceLimitation because we also want to follow canary event
			if err := s.syncBranches(ctx, logger, cluster, newPolicy.PoolName, newPolicy.RuLimitMode); err != nil {
				status = metric.StatusErr
			}
			// suppress any error and continue. use metric to trigger alerts, we need to Add(5) to trigger certainly since the alert rule is
			// for eventbus and it will retry 5 times.
			metric.BranchSyncHandlerCounter.WithLabelValues("sync-tidb-pool", strconv.FormatUint(cluster.ClusterId, 10), constant.EventTypeResourceLimitation, status).
				Add(5)
		}
	}

	err := s.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		clusterEntity, err := s.clusterRepo.GetClusterByID(ctx, record.ClusterID)
		if err != nil {
			return err
		}
		// if the cluster is marked as canary because of project canary, make sure the project still exists
		if record.ProjectID > 0 {
			exists := true
			_, err := s.canaryRepo.GetProject(ctx, record.ProjectID)
			if err != nil {
				if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
					return err
				}
				exists = false
			}
			if exists != record.CanaryMark {
				return nil
			}
		}
		clusterEntity.CanaryMark = record.CanaryMark
		return s.clusterRepo.Save(ctx, clusterEntity)
	})
	if err != nil {
		logger.Error("failed to save cluster entity", zap.Error(err))
		return err
	}
	// should get latest cluster info from mgmt
	record.PoolName = cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoTidbPoolSelector]
	record.RuLimitMode = cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode]
	return s.clusterTidbPoolRepo.MarkAsSynced(ctx, record)
}

func isBillingSpendLimitEvent(status string) bool {
	switch status {
	case constant.SpendLimitEventStatusFit,
		constant.SpendLimitEventStatusExcessive,
		constant.CreditControlEventStatusClosing,
		constant.CreditControlEventStatusRecovered:
		return true
	default:
		return false
	}
}

func ruLimitModeToClusterRUMode(ruLimitMode string) constant.ClusterRUMode {
	if ruLimitMode == constant.RuLimitModeThrottled {
		return constant.ClusterRUModeThrottled
	}
	return constant.ClusterRUModeStandard
}

func (s *SyncTidbPoolService) handleEssentialPlanCluster(
	ctx context.Context,
	logger *zap.Logger,
	record *entity.ClusterTidbPool,
	cluster *tidbmgmtservicev1.Cluster,
) error {
	logger.Info("handle essential plan cluster")

	annotations := cluster.GetServerlessCluster().GetCluster().GetAnnotations()
	ruLimitMode := annotations[constant.AnnoRuLimitMode]
	if ruLimitMode == "" {
		ruLimitMode = constant.RuLimitModeStandard
	}

	if isBillingSpendLimitEvent(record.SpendLimitStatus) {
		plan, err := s.billingFacade.GetTenantPlan(ctx, cluster.OrganizationId)
		if err != nil {
			logger.Error("get tenant plan failed", zap.Error(err))
			return err
		}
		desired := constant.RuLimitModeStandard
		updateCr := make(map[string]string)
		if plan != nil && plan.EffectivePlan == constant.CreditControlEventStatusClosing {
			desired = constant.RuLimitModeThrottled
		}
		if desired != ruLimitMode &&
			(desired == constant.RuLimitModeThrottled || ruLimitMode == constant.RuLimitModeThrottled) {
			updateCr[constant.AnnoRuLimitMode] = desired
		}
		if plan != nil && plan.EffectivePlan != "" && annotations[constant.AnnoTenantPlan] != plan.EffectivePlan {
			updateCr[constant.AnnoTenantPlan] = plan.EffectivePlan
		}
		if len(updateCr) > 0 {
			logger.Info("update essential cluster annotations", zap.Any("body", updateCr))
			if _, err := s.mgmtFacade.UpdateServerlessClusterAnnotations(ctx, cluster.ClusterId, updateCr); err != nil {
				logger.Error("update cluster annotations failed", zap.Error(err))
				return err
			}
			if mode, ok := updateCr[constant.AnnoRuLimitMode]; ok {
				ruLimitMode = mode
				status := metric.StatusOK
				if err := s.syncBranches(ctx, logger, cluster, annotations[constant.AnnoTidbPoolSelector], ruLimitMode); err != nil {
					status = metric.StatusErr
				}
				metric.BranchSyncHandlerCounter.WithLabelValues("sync-tidb-pool", strconv.FormatUint(cluster.ClusterId, 10), constant.EventTypeResourceLimitation, status).
					Add(5)
			}
		}
	}

	clusterEntity := &entity.Cluster{
		ID:                   cluster.ClusterId,
		KeyspaceID:           uint64(cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()),
		TenantID:             cluster.OrganizationId,
		ProjectID:            cluster.ProjectId,
		CloudProviderName:    cluster.CloudProvider,
		RegionName:           cluster.Region,
		CreatedTime:          time.Unix(cluster.GetServerlessCluster().GetCluster().CreateTs, 0),
		CreationFinishedTime: nil,
		SpendLimit:           decimal.NewFromInt(-1), // essential plan cluster has no Spending Limit
		FreeCredit:           decimal.NewFromInt(-1), // essential plan cluster has no free credit
		IsPaid:               true,
		CanaryMark:           annotations[constant.AnnoCanary] == constant.CanaryPoolMarkedFlag,
		RuMode:               ruLimitModeToClusterRUMode(ruLimitMode),
	}
	if createdAt := cluster.GetServerlessCluster().GetCreationFinishedAt(); createdAt > 0 {
		t := time.Unix(createdAt, 0)
		clusterEntity.CreationFinishedTime = &t
	}

	if err := s.clusterRepo.Save(ctx, clusterEntity); err != nil {
		logger.Error("save essential plan cluster failed", zap.Error(err))
		return err
	}
	record.PoolName = annotations[constant.AnnoTidbPoolSelector]
	record.RuLimitMode = ruLimitMode
	record.ProjectID = cluster.ProjectId

	return s.clusterTidbPoolRepo.MarkAsSynced(ctx, record)
}

func (s *SyncTidbPoolService) handleStarterPlanCluster(
	ctx context.Context,
	logger *zap.Logger,
	record *entity.ClusterTidbPool,
	cluster *tidbmgmtservicev1.Cluster,
) error {
	logger.Info("handle starter plan cluster")
	/* step 1: gen schedule policy according to billing spending limit config */
	setting, berr := s.billingFacade.GetSpendLimit(ctx, cluster.OrganizationId, cluster.ClusterId)
	if berr != nil {
		return berr
	}
	bizData := &facade.ServerlessSpendLimitBizData{}
	if err := json.Unmarshal([]byte(setting.BizData), bizData); err != nil {
		return err
	}

	info, berr := s.billingFacade.GetSpendLimitInfo(ctx, setting)
	if berr != nil {
		return berr
	}
	newPolicy := s.genTidbPoolPolicy(setting, info, bizData)
	newPolicy, err := s.reEvaluatePolicy(newPolicy, info)
	if err != nil {
		return err
	}
	newPolicy = s.reEvaluatePoolPolicy(newPolicy, cluster)

	/* step 2: rewrite schedule policy according to canary mark */
	if record.SpendLimitStatus != constant.SpendLimitEventStatusCreateCluster {
		c, err := s.clusterRepo.GetClusterByID(ctx, cluster.ClusterId)
		if err != nil {
			return err
		}
		if c.CanaryMark {
			newPolicy.PoolName = constant.IntoCanaryPool(newPolicy.PoolName)
			newPolicy.CanaryMark = constant.CanaryPoolMarkedFlag
		}
	}

	/* step 3: update cluster annotations only if the policy has been changed */
	originPolicy := &tidbPoolPolicy{
		PoolName:    cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoTidbPoolSelector],
		RuLimitMode: cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode],
		CanaryMark:  cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoCanary],
	}
	updateCr := make(map[string]string)
	if originPolicy.PoolName != newPolicy.PoolName {
		updateCr[constant.AnnoTidbPoolSelector] = newPolicy.PoolName
	}
	if originPolicy.RuLimitMode != newPolicy.RuLimitMode {
		updateCr[constant.AnnoRuLimitMode] = newPolicy.RuLimitMode
	}
	if originPolicy.CanaryMark != newPolicy.CanaryMark {
		updateCr[constant.AnnoCanary] = newPolicy.CanaryMark
	}
	if len(updateCr) > 0 {
		updateCr[constant.AnnoTenantPlan] = info.Status
		logger.Info("update cluster annotations", zap.Any("body", updateCr))
		if _, err = s.mgmtFacade.UpdateServerlessClusterAnnotations(ctx, cluster.ClusterId, updateCr); err != nil {
			return err
		}
	}
	record.PoolName = newPolicy.PoolName
	record.RuLimitMode = newPolicy.RuLimitMode
	record.ProjectID = cluster.ProjectId

	/* step 4: if create cluster event, save cluster entity and return */
	if record.SpendLimitStatus == constant.SpendLimitEventStatusCreateCluster {
		logger.Info("handle create cluster event")
		clusterEntity := &entity.Cluster{
			ID:                   cluster.ClusterId,
			KeyspaceID:           uint64(cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()),
			TenantID:             cluster.OrganizationId,
			ProjectID:            cluster.ProjectId,
			CloudProviderName:    cluster.CloudProvider,
			RegionName:           cluster.Region,
			CreatedTime:          time.Unix(cluster.GetServerlessCluster().GetCluster().CreateTs, 0),
			CreationFinishedTime: nil,
			SpendLimit:           bizData.InvoiceLimit,
			FreeCredit:           bizData.FreeCredits,
			IsPaid:               bizData.InvoiceLimit.GreaterThan(decimal.Zero),
			CanaryMark:           cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoCanary] == constant.CanaryPoolMarkedFlag,
			RuMode:               getClusterRuModeFromPolicy(newPolicy),
		}
		if cluster.GetServerlessCluster().CreationFinishedAt != nil {
			t := time.Unix(*cluster.GetServerlessCluster().CreationFinishedAt, 0)
			clusterEntity.CreationFinishedTime = &t
		}
		// upsert cluster with attribute `is_paid` after setting/modifying Spending Limit
		if err := s.clusterRepo.Save(ctx, clusterEntity); err != nil {
			return err
		}
	} else {
		clusterEntity, err := s.clusterRepo.GetClusterByID(ctx, record.ClusterID)
		if err != nil {
			return err
		}
		clusterEntity.SpendLimit = bizData.InvoiceLimit
		clusterEntity.FreeCredit = bizData.FreeCredits
		clusterEntity.IsPaid = bizData.InvoiceLimit.GreaterThan(decimal.Zero)
		if err := s.clusterRepo.Save(ctx, clusterEntity); err != nil {
			return err
		}
	}

	/* step 5: if billing spending limit events, notify eventbus and lark group */
	logger.Info("handle billing spending limit events")
	if len(updateCr) > 0 {
		/* notify eventbus if the ru limit mode has been changed */
		if originPolicy.RuLimitMode != newPolicy.RuLimitMode {
			if s.eventbusService != nil {
				if err := s.eventbusService.Publish(&tieventbus.Event{
					ID:   shortuuid.New(),
					Type: event.DisplayEventTopic,
					Data: event.DisplayEvent{
						DisplayEventType: constant.EventTypeResourceLimitation,
						DisplayStatus:    event.EventStatusSucceeded,
						Severity:         event.EventSeverityInfo,
						Summary: fmt.Sprintf(constant.EventSummaryResourceLimitation,
							constant.EventSummaryRUMode[newPolicy.RuLimitMode],
							ruModeChangeDesc(originPolicy.RuLimitMode, newPolicy.RuLimitMode),
							constant.EventDetailsSpendLimitEvents[newPolicy.SpendLimitStatus],
							constant.EventDetailsResourceLimitation[newPolicy.PoolName+newPolicy.RuLimitMode],
						),
						HappenedAtMs: uint64(time.Now().UTC().UnixMilli()),
						AuthorID:     0,
						ClusterID:    cluster.ClusterId,
						ProjectID:    cluster.ProjectId,
						TenantID:     cluster.OrganizationId,
						Detail:       "event_type: " + newPolicy.SpendLimitStatus,
					},
				}); err != nil {
					return err
				}
			}
			if err := s.clusterRepo.UpdateRUMode(ctx, cluster.ClusterId, getClusterRuModeFromPolicy(newPolicy)); err != nil {
				return err
			}
		}
		/* notify lark group for new schedule policy, only for billing spending limit events */
		if (originPolicy.PoolName != newPolicy.PoolName ||
			originPolicy.RuLimitMode != newPolicy.RuLimitMode) &&
			!utils.IsNotifyBlockTenantsAndEnv(cluster.OrganizationId) {
			title := fmt.Sprintf(
				"[%s] Update Tidb Schedule Policy",
				strings.ToUpper(config.Global().ServerConfig.Stage),
			)
			tenant, _ := s.accountFacade.GetTenant(ctx, cluster.OrganizationId)
			if tenant == nil {
				tenant = &facade.Tenant{}
			}
			message := fmt.Sprintf("Tenant: %d / %s / %s / %v\n"+
				"Cluster: %d/%s / %s/%s\n"+
				"EventType: %s\n"+
				"EventStatus: %s\n"+
				"OldPolicy: %s / %s\n"+
				"NewPolicy: %s / %s\n",
				cluster.OrganizationId, tenant.Name, tenant.Email, tenant.Kind,
				cluster.ClusterId, cluster.ClusterName, cluster.CloudProvider, cluster.Region,
				record.Type,
				record.SpendLimitStatus,
				originPolicy.PoolName,
				originPolicy.RuLimitMode,
				newPolicy.PoolName,
				newPolicy.RuLimitMode,
			)
			go func() {
				utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
			}()
		}
		/* sync branches */
		{
			status := metric.StatusOK
			// can't solely rely on EventTypeResourceLimitation because we also want to follow canary event
			if err := s.syncBranches(ctx, logger, cluster, newPolicy.PoolName, newPolicy.RuLimitMode); err != nil {
				status = metric.StatusErr
			}
			// suppress any error and continue. use metric to trigger alerts, we need to Add(5) to trigger certainly since the alert rule is
			// for eventbus and it will retry 5 times.
			metric.BranchSyncHandlerCounter.WithLabelValues("sync-tidb-pool", strconv.FormatUint(cluster.ClusterId, 10), constant.EventTypeResourceLimitation, status).
				Add(5)
		}
	}
	return s.clusterTidbPoolRepo.MarkAsSynced(ctx, record)
}

func (s *SyncTidbPoolService) handleCanaryEventImpl(
	ctx context.Context,
	record *entity.ClusterTidbPool,
) (exitReason string, err error) {
	cluster, err := s.clusterRepo.GetClusterByID(ctx, record.ClusterID)
	if err != nil {
		exitReason = exitReasonFailToGetClusterById
		return exitReason, err
	}
	// if the cluster is marked as canary because of project canary, make sure the project still exists
	if record.ProjectID > 0 {
		exists := true
		_, err := s.canaryRepo.GetProject(ctx, record.ProjectID)
		if err != nil {
			if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
				exitReason = "failed to get project canary by id"
				return exitReason, err
			}
			exists = false
		}
		if exists != record.CanaryMark {
			return "", nil
		}
	}
	cluster.CanaryMark = record.CanaryMark
	err = s.clusterRepo.Save(ctx, cluster)
	if err != nil {
		exitReason = "failed to save cluster entity"
		return exitReason, err
	}

	return "", nil
}

func (s *SyncTidbPoolService) handleCanaryEvent(
	ctx context.Context,
	record *entity.ClusterTidbPool,
	policy *tidbPoolPolicy,
) (_ *tidbPoolPolicy, exitReason string, err error) {
	if record.CanaryMark {
		policy.PoolName = constant.IntoCanaryPool(policy.PoolName)
		policy.CanaryMark = constant.CanaryPoolMarkedFlag
	} else {
		policy.CanaryMark = constant.CanaryPoolUnmarkedFlag
	}

	err = s.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		exitReason, err = s.handleCanaryEventImpl(ctx, record)
		return err
	})
	if err != nil {
		return nil, exitReason, err
	}

	return policy, "", nil
}

func (s *SyncTidbPoolService) syncOneCluster(ctx context.Context, record *entity.ClusterTidbPool) (err error) {
	status, exitReason := metric.StatusOK, "-"
	prevPoolName, prevRuLimitMode := "", ""
	poolName, ruLimitMode := "", ""
	needSendLarkMsg := false
	var cluster *tidbmgmtservicev1.Cluster
	logger := utils.GetWorkerLoggerFromContext(ctx).With(zap.Uint64("cluster_id", record.ClusterID),
		zap.String("event_type", record.Type), zap.String("event_status", record.SpendLimitStatus))
	ctx, span := Tracer.Start(ctx, "synctidbpool.syncOneCluster")
	defer func() {
		tracing.EndSpan(span, err)
		if err != nil {
			needSendLarkMsg = true
			status = metric.StatusErr + ", " + exitReason
			logger.Error(exitReason, zap.Error(err))
		}

		if record.Type == constant.TypeCanary && exitReason == exitReasonFailToGetClusterById {
			// Disable lark alert for canary event if the clusterRepo is not synced by Spending Limit event.
			needSendLarkMsg = false
		}

		metric.TidbScheduleEventCounter.WithLabelValues(
			fmt.Sprint(record.ClusterID),
			"sync-record",
			record.SpendLimitStatus,
			prevPoolName+"/"+prevRuLimitMode,
			poolName+"/"+ruLimitMode,
			status,
			exitReason,
		).Inc()

		if !needSendLarkMsg {
			return
		}

		if cluster == nil {
			return
		}

		// send feishu message
		logger.Info(
			"to send feishu notification",
			zap.String("prev_pool", prevPoolName),
			zap.String("new_pool", poolName),
		)
		if record.SpendLimitStatus != constant.SpendLimitEventStatusHandy &&
			!utils.IsNotifyBlockTenantsAndEnv(cluster.OrganizationId) {
			title := fmt.Sprintf(
				"[%s] Update Tidb Schedule Policy",
				strings.ToUpper(config.Global().ServerConfig.Stage),
			)
			tenant, _ := s.accountFacade.GetTenant(ctx, cluster.OrganizationId)
			if tenant == nil {
				tenant = &facade.Tenant{}
			}
			message := fmt.Sprintf("Tenant: %d / %s / %s / %v\n"+
				"Cluster: %d/%s / %s/%s\n"+
				"EventType: %s\n"+
				"EventStatus: %s\n"+
				"OldPolicy: %s / %s\n"+
				"NewPolicy: %s / %s\n"+
				"Result: %s",
				cluster.OrganizationId, tenant.Name, tenant.Email, tenant.Kind,
				cluster.ClusterId, cluster.ClusterName, cluster.CloudProvider, cluster.Region,
				record.Type,
				record.SpendLimitStatus,
				prevPoolName,
				prevRuLimitMode,
				poolName,
				ruLimitMode,
				status)
			go func() {
				utils.SendLarkMessage(title, message, utils.LarkMsgLevelInfo)
			}()
		}
	}()

	logger.Info("sync tidb pool start.")
	span.AddEvent("get serverless cluter from mgmt-svc")
	cluster, berr := s.mgmtFacade.GetCluster(ctx, record.ClusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if berr != nil {
		if berr.RawErrCode() == bizerr.ErrCodeRecordNotFound {
			if e := s.clusterTidbPoolRepo.RemoveAllEvents(ctx, record.ClusterID); e != nil {
				berr = e
				exitReason = "failed to remove deleted cluster from db"
			}
		} else {
			exitReason = "failed to get cluster from mgmt-svc"
		}
		return berr
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		exitReason = "cluster is not running"
		return nil
	}
	annotations := cluster.GetServerlessCluster().GetCluster().GetAnnotations()
	if annotations == nil {
		exitReason = "cluster annotations is null"
		return nil
	}
	/*
		rcu > 0, essential plan, without Spending Limit config
	*/
	if cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu() > 0 {
		return s.handleEssentialPlanCluster(ctx, logger, record, cluster)
	}
	span.AddEvent("get Spending Limit setting")
	setting, err := s.billingFacade.GetSpendLimit(ctx, cluster.OrganizationId, cluster.ClusterId)
	if err != nil {
		exitReason = "fail to get Spending Limit setting from billing"
		return err
	}
	bizData := &facade.ServerlessSpendLimitBizData{}
	if err := json.Unmarshal([]byte(setting.BizData), bizData); err != nil {
		exitReason = "fail to get Spending Limit bizData from billing"
		return err
	}

	info, err := s.billingFacade.GetSpendLimitInfo(ctx, setting)
	if err != nil {
		exitReason = "fail to get Spending Limit info from billing"
		return err
	}

	// generate tidb pool policy
	policy := s.genTidbPoolPolicy(setting, info, bizData)
	policy, err = s.reEvaluatePolicy(policy, info)
	if err != nil {
		exitReason = "failed to re-evaluate schedule policy"
		return err
	}

	switch record.Type {
	case constant.TypeSpendLimit:
		if !slices.Contains(constant.SpendLimitValidStatuses, record.SpendLimitStatus) {
			exitReason = "invalid Spending Limit status"
			err = errors.New(exitReason)
			goto RET
		}
		policy, exitReason, err = s.handleSpendLimitEvent(ctx, record, cluster, policy, bizData)
	case constant.TypeCanary:
		policy, exitReason, err = s.handleCanaryEvent(ctx, record, policy)
	default:
		exitReason = "unknown event type"
		err = errors.New(exitReason)
		goto RET
	}
	if err != nil {
		return err
	}

	poolName = policy.PoolName
	ruLimitMode = policy.RuLimitMode
	prevPoolName = annotations[constant.AnnoTidbPoolSelector]
	prevRuLimitMode = annotations[constant.AnnoRuLimitMode]
	/*
		if constant.DisableScheduleWhiteList[record.ClusterID] {
			logger.Info("disable to schedule tidb pool",
				zap.Uint64("cluster", record.ClusterID),
				zap.String("from", fmt.Sprintf("%s/%s", prevPoolName, prevRuLimitMode)),
				zap.String("target to", fmt.Sprintf("%s/%s", poolName, ruLimitMode)))
			goto RET
		}
	*/

	if prevPoolName != poolName || prevRuLimitMode != ruLimitMode {
		needSendLarkMsg = true
		update := map[string]string{
			constant.AnnoTidbPoolSelector: poolName,
			constant.AnnoRuLimitMode:      ruLimitMode,
			constant.AnnoCanary:           policy.CanaryMark,
		}
		record.RuLimitMode = ruLimitMode
		record.PoolName = poolName
		span.AddEvent("update cluster annotations")
		if _, err = s.mgmtFacade.UpdateServerlessClusterAnnotations(ctx, cluster.ClusterId, update); err != nil {
			exitReason = "failed to update cluster annotations"
			return err
		}
		// exclude events triggered by cluster creation and canary event
		if s.eventbusService != nil && prevRuLimitMode != "" && record.Type != constant.TypeCanary {
			err := s.eventbusService.Publish(&tieventbus.Event{
				ID:   shortuuid.New(),
				Type: event.DisplayEventTopic,
				Data: event.DisplayEvent{
					DisplayEventType: constant.EventTypeResourceLimitation,
					DisplayStatus:    event.EventStatusSucceeded,
					Severity:         event.EventSeverityInfo,
					Summary: fmt.Sprintf(constant.EventSummaryResourceLimitation,
						constant.EventSummaryRUMode[ruLimitMode],
						ruModeChangeDesc(prevRuLimitMode, ruLimitMode),
						constant.EventDetailsSpendLimitEvents[policy.SpendLimitStatus],
						constant.EventDetailsResourceLimitation[policy.OriginPoolName+ruLimitMode],
					),
					HappenedAtMs: uint64(time.Now().UTC().UnixMilli()),
					AuthorID:     0,
					ClusterID:    cluster.ClusterId,
					ProjectID:    cluster.ProjectId,
					TenantID:     cluster.OrganizationId,
					Detail:       "event_type: " + policy.SpendLimitStatus,
				},
			})
			if err != nil {
				logger.Error("failed to publish events center", zap.Error(err))
				return err
			}
		}
		logger.Info("update tidb pool policy",
			zap.String("from", fmt.Sprintf("%s/%s", prevPoolName, prevRuLimitMode)),
			zap.String("to", fmt.Sprintf("%s/%s", poolName, ruLimitMode)))

		// exclude events triggered by cluster creation
		if prevRuLimitMode != "" {
			span.AddEvent("update child branches")
			var status string
			// can't solely rely on EventTypeResourceLimitation because we also want to follow canary event
			if err := s.syncBranches(ctx, logger, cluster, poolName, ruLimitMode); err != nil {
				status = metric.StatusErr
			} else {
				status = metric.StatusOK
			}
			// suppress any error and continue. use metric to trigger alerts, we need to Add(5) to trigger certainly since the alert rule is
			// for eventbus and it will retry 5 times.
			metric.BranchSyncHandlerCounter.WithLabelValues("sync-tidb-pool", strconv.FormatUint(cluster.ClusterId, 10), constant.EventTypeResourceLimitation, status).
				Add(5)
		}
	}

RET:
	span.AddEvent("mark as synced")
	if e := s.clusterTidbPoolRepo.MarkAsSynced(ctx, record); e != nil {
		exitReason = "failed to mark as synced to db"
		return e
	}

	logger.Info("sync tidb pool succeed")
	return err
}

type tidbPoolPolicy struct {
	PoolName         string
	RuLimitMode      string
	SpendLimitStatus string
	OriginPoolName   string
	CanaryMark       string
}

func getClusterRuModeFromPolicy(policy *tidbPoolPolicy) constant.ClusterRUMode {
	if policy.RuLimitMode == constant.RuLimitModeThrottled {
		return constant.ClusterRUModeThrottled
	}
	return constant.ClusterRUModeStandard
}

func (s *SyncTidbPoolService) genTidbPoolPolicy(
	setting *facade.SpendLimitSetting,
	spendLimitInfo *facade.SpendLimitInfo,
	bizData *facade.ServerlessSpendLimitBizData,
) *tidbPoolPolicy {
	spendLimitStatus := constant.SpendLimitEventStatusFit
	if spendLimitInfo.Costs.GreaterThanOrEqual(setting.Limit) {
		spendLimitStatus = constant.SpendLimitEventStatusExcessive
	}

	var poolName, ruLimitMode, originPoolName string
	if bizData.InvoiceLimit.GreaterThan(decimal.Zero) {
		/* for Starter/Scalable clusters */
		poolName = constant.TidbPoolNameVip
		originPoolName = constant.TidbPoolNameVip
		ruLimitMode = constant.RuLimitModeStandard
		if spendLimitStatus == constant.SpendLimitEventStatusExcessive {
			ruLimitMode = constant.RuLimitModeThrottled
		}
	} else {
		/* for Starter/Free clusters */
		poolName = constant.TidbPoolNameFree
		originPoolName = constant.TidbPoolNameFree
		ruLimitMode = constant.RuLimitModeStandard
		if spendLimitStatus == constant.SpendLimitEventStatusExcessive {
			ruLimitMode = constant.RuLimitModeThrottled
		}
	}
	return &tidbPoolPolicy{
		PoolName:         poolName,
		RuLimitMode:      ruLimitMode,
		SpendLimitStatus: spendLimitStatus,
		OriginPoolName:   originPoolName,
	}
}

func (s *SyncTidbPoolService) reEvaluatePolicy(
	policy *tidbPoolPolicy,
	spendLimitInfo *facade.SpendLimitInfo,
) (*tidbPoolPolicy, error) {
	if policy == nil || spendLimitInfo == nil {
		return nil, errors.New("policy or spendLimitInfo is nil")
	}
	if spendLimitInfo.Status == constant.CreditControlEventStatusClosing {
		policy.RuLimitMode = constant.RuLimitModeThrottled
	}
	return policy, nil
}

func (s *SyncTidbPoolService) reEvaluatePoolPolicy(
	policy *tidbPoolPolicy,
	cluster *tidbmgmtservicev1.Cluster,
) *tidbPoolPolicy {
	if cluster.GetOrganizationId() == 1372813089209302377 &&
		cluster.GetProjectId() == 1372813089454620624 &&
		policy != nil {
		// 1372813089209302377/1372813089454620624 is a dedicated tenant/project for zero plan clusters
		// schedule all the related starter clusters into free pool by default
		policy.PoolName = constant.TidbPoolNameFree
	}
	return policy
}

func ruModeChangeDesc(prevRuLimitMode, ruLimitMode string) string {
	if prevRuLimitMode == "" || prevRuLimitMode == ruLimitMode {
		return ruLimitMode
	}
	return fmt.Sprintf("%s -> %s", prevRuLimitMode, ruLimitMode)
}

func (s *SyncTidbPoolService) syncBranches(
	ctx context.Context,
	logger *zap.Logger,
	cluster *tidbmgmtservicev1.Cluster,
	pool string,
	rumode string,
) error {
	branches, err := s.branchRepo.ListBranchesByMainBranchID(ctx, strconv.FormatUint(cluster.ClusterId, 10), nil)
	if err != nil {
		logger.Error("list branches by root branch id failed", zap.Error(err))
		return err
	}
	if len(branches) == 0 {
		logger.Info("no child branches")
		return nil
	}

	for _, branch := range branches {
		branLog := logger.With(zap.String("branch_id", branch.ID))
		br, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
		if err != nil {
			branLog.Error("get branch CR failed", zap.Error(err))
			return err
		}
		if br.Status.Phase != cloudv1.BranchPhaseRunning {
			branLog.Info("child branch is not running, skip update")
			continue
		}

		needUpdate := false
		if br.Annotations[constant.AnnoTidbPoolSelector] != pool {
			branLog = branLog.With(
				zap.String("prev_tidb_pool_selector", br.Annotations[constant.AnnoTidbPoolSelector]),
				zap.String("later_tidb_pool_selector", pool),
			)
			br.Annotations[constant.AnnoTidbPoolSelector] = pool
			needUpdate = true
		}

		// Note: limiter may override normal->throttled based on size on next tick.
		if br.Annotations[constant.AnnoRuLimitMode] != rumode {
			branLog = branLog.With(zap.String("prev_ru_limit_mode", br.Annotations[constant.AnnoRuLimitMode]),
				zap.String("later_ru_limit_mode", rumode))
			br.Annotations[constant.AnnoRuLimitMode] = rumode
			needUpdate = true
		}

		if needUpdate {
			err := s.infraFacade.UpdateBranch(ctx, br)
			if err != nil {
				branLog.Error("update child branch CR failed", zap.Error(err))
				return err
			}
			branLog.Info("update child branch CR succeed")
		} else {
			branLog.Info("no need to update child branch CR")
		}
	}

	logger.Info("update all child branches")
	return nil
}