package service

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

	"github.com/dustin/go-humanize"
	"github.com/lithammer/shortuuid/v3"
	core_event "github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/pkgv2/log"
	"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/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"github.com/tidbcloud/serverless-service/pkg/event"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"gorm.io/datatypes"
	api_errors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/util/retry"
	"k8s.io/utils/pointer"
)

const (
	reservedBranchName = "main"
)

var backoff = wait.Backoff{
	Steps:    5,
	Duration: 1 * time.Second,
	Factor:   2,
	Jitter:   0.1,
}

type BranchService interface {
	CreateBranch(
		ctx context.Context,
		clusterId, branchName, parentID string,
		parentTimestamp *time.Time,
		rootPassword string,
		userID uint64,
		annotations map[string]string,
	) (*vo.Branch, bizerr.BizError)
	ListBranchesInPage(
		ctx context.Context,
		mainBranchID string,
		opt vo.ListBranchesOpt,
	) ([]*vo.Branch, int64, bizerr.BizError)
	ListBranches(ctx context.Context, mainBranchID string, deleteAfterTime *time.Time) ([]*vo.Branch, bizerr.BizError)
	GetBranchById(ctx context.Context, mainBranchID string, branchId string) (*vo.Branch, bizerr.BizError)
	DeleteBranchById(
		ctx context.Context,
		mainBranchID string,
		branchId string,
		userID uint64,
	) (*vo.Branch, bizerr.BizError)
	UpdateBranchHasSetPwd(ctx context.Context, branchId string, hasSetPwd bool) (bizErr bizerr.BizError)
	UpdateBranch(
		ctx context.Context,
		mainBranchID string,
		branchId string,
		branch *vo.UpdateBranchReq,
	) (*vo.Branch, bizerr.BizError)
	CollectMetricsFromDB(ctx context.Context) bizerr.BizError
	ListBranchesRowStorage(ctx context.Context, branches []*vo.Branch) (map[string]float64, error)
	DeleteBranchByClusterId(ctx context.Context, clusterId string, userID uint64) bizerr.BizError
	ResetBranchById(ctx context.Context, mainBranchID string, branchId string) (*vo.Branch, bizerr.BizError)
	UpdateSqlUser(ctx context.Context, clusterId, branchId, username, password string) bizerr.BizError
}

type branchService struct {
	infraFacade                facade.InfraFacade
	branchRepo                 repo.BranchRepo
	o11yFacade                 facade.O11yFacade
	eventbusService            *eventbus.EventbusService
	lockRepo                   repo.LockRepo
	transProvider              db.TransProvider
	mgmtSvc                    facade.MgmtSvcFacade
	billingFacade              facade.BillingFacade
	encryptorFacade            facade.EncryptorFacade
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
}

func NewBranchService(
	infraFacade facade.InfraFacade,
	branchRepo repo.BranchRepo,
	o11yFacade facade.O11yFacade,
	eventbusService *eventbus.EventbusService,
	lockRepo repo.LockRepo,
	transProvider db.TransProvider,
	mgmtSvc facade.MgmtSvcFacade,
	billingFacade facade.BillingFacade,
	encryptorFacade facade.EncryptorFacade,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
) BranchService {
	return &branchService{
		infraFacade:                infraFacade,
		branchRepo:                 branchRepo,
		o11yFacade:                 o11yFacade,
		eventbusService:            eventbusService,
		lockRepo:                   lockRepo,
		transProvider:              transProvider,
		mgmtSvc:                    mgmtSvc,
		billingFacade:              billingFacade,
		encryptorFacade:            encryptorFacade,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
	}
}

func (s *branchService) CreateBranch(
	ctx context.Context,
	clusterId, branchName, parentID string,
	parentTimestamp *time.Time,
	rootPassword string,
	userID uint64,
	annotations map[string]string,
) (_ *vo.Branch, bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("CreateBranch", bizErr)
	}()
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId), zap.String("branch_name", branchName), zap.String("parent_id", parentID))

	if branchName == "" || len(branchName) > 64 {
		logger.Info("invalid branch name length, should be with 1-64, inclusive")
		return nil, bizerr.New(errorcode.ErrCodeBranchNameInvalidLength, "")
	}
	if strings.EqualFold(branchName, reservedBranchName) {
		logger.Info("reserved name is not allowed, please try another name")
		return nil, bizerr.New(errorcode.ErrCodeBranchNameReserved, "")
	}

	if !utils.IsBranch(parentID) {
		return s.createBranchFromCluster(ctx, branchName, parentID, parentTimestamp, rootPassword, userID, annotations)
	}

	return s.createBranchFromBranch(
		ctx,
		clusterId,
		branchName,
		parentID,
		parentTimestamp,
		rootPassword,
		userID,
		annotations,
	)
}

func (s *branchService) createBranchFromCluster(
	ctx context.Context,
	branchName string,
	parentClusterID string,
	parentTimestamp *time.Time,
	rootPassword string,
	userID uint64,
	annotations map[string]string,
) (*vo.Branch, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("branch_name", branchName), zap.String("cluster_id", parentClusterID))
	// generate the branch id up front so every log line, including the
	// validation-step timings, carries branch_id
	// k8s resource name must be lowercase. https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-subdomain-names
	id := entity.GenerateIDWithPrefix(constant.BranchIDPrefix)
	logger = logger.With(zap.String("branch_id", id))
	funcStart := time.Now()

	clusterID, err := strconv.ParseUint(parentClusterID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "invalid cluster id %s", parentClusterID)
	}

	stepStart := time.Now()
	parentCluster, bizErr := s.mgmtSvc.GetClusterIncludeDedicated(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_FULL,
	)
	logger.Info("[branch lifecycle] branch create step",
		zap.String("step", "mgmt_get_cluster"),
		zap.Int64("duration_ms", time.Since(stepStart).Milliseconds()))
	if bizErr != nil {
		logger.Error("failed to get cluster from mgmt svc", zap.Error(bizErr))
		return nil, bizErr
	}
	if parentCluster.DeployType != tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS {
		logger.Error(
			"only support create branch from serverless cluster",
			zap.Int("deploy_type", int(parentCluster.DeployType)),
		)
		return nil, bizerr.New(errorcode.ErrCodeCreateBranchNotFromServerlessCluster, "")
	}
	if parentCluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		logger.Error(
			"only support create branch from normal cluster",
			zap.Int("status", int(parentCluster.GetServerlessCluster().GetUserStatus())),
		)
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "")
	}
	if parentCluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode] == constant.RuLimitModeThrottled {
		return nil, bizerr.New(errorcode.ErrCodeCreateBranchFromThrottledCluster, "")
	}

	bizErr = s.exceedNumberLimit(ctx, parentCluster.OrganizationId)
	if bizErr != nil {
		logger.Error("failed to check exceed number limit", zap.Error(bizErr))
		return nil, bizErr
	}

	paid := parentCluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu() > 0 ||
		parentCluster.GetServerlessCluster().GetSpendLimit().GetMonthly() > 0
	// Predict only paid clusters might reach the limit.
	if config.Global().BizConfig.BranchParentSizeUpperLimitBytes != 0 && paid {
		stepStart = time.Now()
		storage, err := s.o11yFacade.GetClusterRowStorage(ctx, parentCluster)
		logger.Info("[branch lifecycle] branch create step",
			zap.String("step", "o11y_get_row_storage"),
			zap.Int64("duration_ms", time.Since(stepStart).Milliseconds()))
		if err != nil {
			logger.Error(
				"failed to get cluster storage size from o11y",
				zap.Error(err),
				zap.Uint64("cluster_id", parentCluster.ClusterId),
			)
			return nil, bizerr.New(errorcode.ErrCodeSvcO11y, errorcode.ErrMsgGetSkuUsagesFailed)
		}
		if storage > config.Global().BizConfig.BranchParentSizeUpperLimitBytes {
			if _, ok := config.Global().BizConfig.BranchTenantNumberQuota[parentCluster.OrganizationId]; ok {
				logger.Info(
					"tenant found in branch tenant number quota, won't limit it",
					zap.Uint64("tenant_id", parentCluster.OrganizationId),
				)
			} else {
				return nil, bizerr.New(errorcode.ErrCodeParentBranchSizeTooLarge,
					"Creating branches from a cluster larger than %s is not supported. Contact us for larger size.",
					humanize.IBytes(uint64(config.Global().BizConfig.BranchParentSizeUpperLimitBytes)))
			}
		}
	}

	// set annotation
	annotationJson, err := marshalAnnotations(annotations)
	if err != nil {
		logger.Error("failed to marshal annotations", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInvalidBranchAnnotation, "")
	}

	salt, err := entity.GenerateSalt()
	if err != nil {
		logger.Error("failed to generate branch salt", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	hasSetRootPwd := false
	if rootPassword != "" {
		encryptedRootPassword, err := s.encryptorFacade.EncryptBase64(rootPassword)
		if err != nil {
			logger.Error("failed to encrypt root password", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		rootPassword = encryptedRootPassword
		hasSetRootPwd = true
	}

	// branches.createdAt is of type TIMESTAMP. It defaults to second precision. If we don't truncate, then
	// branch.MetadataOfBranch will have more precision than the final entity, leading to playground password
	// calculation failed. (2023-04-04T12:46:56.921268434Z vs 2023-04-04T12:46:57Z)
	// Though playground password doesn't exist any more. Still truncate to make it sync.
	now := time.Now().Truncate(time.Second)
	if parentTimestamp != nil {
		// parent timestamp need to truncate to seconds precision
		truncateParentTimestamp := parentTimestamp.Truncate(time.Second)
		parentTimestamp = &truncateParentTimestamp
	}
	backupRetention := parentCluster.GetServerlessCluster().GetBackupPolicy().GetRetentionDays()
	parentTimestamp, err = verifyParentTimestamp(
		parentTimestamp,
		parentCluster.CreateTime.AsTime(),
		backupRetention,
		now,
	)
	if err != nil {
		logger.Error("failed to verify parent timestamp", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInvalidParentTimestamp, "%s", err.Error())
	}
	parentTimestamp = parentTimestampOrDefault(
		parentTimestamp,
		config.Global().BizConfig.BranchOmitParentTimestampKuberneteses,
		parentCluster.GetServerlessCluster().GetCluster().GetKubernetes(),
		now,
	)
	infraProjectId := parentCluster.GetOriginalProjectId()
	xProjectId := parentCluster.ProjectId
	b := &entity.Branch{
		ID:                    id,
		CreatedAt:             now,
		UpdatedAt:             &now,
		DeletedTime:           nil,
		CreationFinishedTime:  nil,
		TenantID:              parentCluster.OrganizationId,
		ProjectID:             infraProjectId,
		CreatedByID:           userID,
		CloudProviderName:     parentCluster.CloudProvider,
		RegionName:            parentCluster.Region,
		Kubernetes:            parentCluster.GetServerlessCluster().GetCluster().GetKubernetes(),
		Name:                  branchName,
		Status:                constant.BranchStatusPending,
		ParentID:              parentClusterID,
		ParentTimestamp:       parentTimestamp,
		MainBranchID:          parentClusterID,
		Salt:                  salt,
		Annotations:           annotationJson,
		CreatedBy:             buildCreatedBy(ctx, annotations),
		EncryptedRootPassword: rootPassword,
		HasSetPwd:             hasSetRootPwd,
	}

	branchCR := &cloudv1.Branch{
		ObjectMeta: metav1.ObjectMeta{
			Annotations: copyAnnotations(ctx, parentCluster),
			Labels:      copyLabels(ctx, parentCluster),
			Namespace:   s.infraFacade.Namespace(infraProjectId),
			Name:        id,
		},
		Spec: cloudv1.BranchSpec{
			SqlEndpoints: cloudv1.BranchSqlEndpointsSpec{
				Public: cloudv1.BranchPublicSpec{
					Service: cloudv1.Service{
						Enabled: pointer.Bool(
							!parentCluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().GetDisabled(),
						),
					},
					AuthorizedNetworks: ConvertRegionalToInfraAPIAuthorizedNetworks(
						parentCluster.GetServerlessCluster().
							GetCluster().
							GetEndpoints().
							GetPublic().
							GetAuthorizedNetwork(),
					),
				},
				PrivateLinkService: &cloudv1.PrivateLinkService{
					AWS: &cloudv1.AWSPrivateLinkService{AcceptedEndpoints: []string{"*"}},
				},
			},
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{Replicas: 1},
				PD:   cloudv1.PD{Replicas: 1},
				TiKV: cloudv1.TiKV{Replicas: 1},
			},
			AutoScaling: &cloudv1.AutoScaling{
				MinRcu: parentCluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu(),
				MaxRcu: parentCluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu(),
			},
			RecoveryMode: false,
		},
	}

	j, err := json.Marshal(b.MetadataOfBranch())
	if err != nil {
		logger.Error("Can't marshal branch metadata", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "")
	}
	branchCR.Annotations[constant.BranchMetadataAnnKey] = string(j)
	branchCR.Annotations[constant.ParentBranchIdAnnKey] = strconv.FormatUint(parentCluster.ClusterId, 10)
	branchCR.Annotations[constant.MainBranchIdAnnKey] = strconv.FormatUint(parentCluster.ClusterId, 10)
	branchCR.Annotations[constant.ParentTypeAnnKey] = constant.ParentTypeCluster
	setParentTimestampAnnotation(branchCR.Annotations, parentTimestamp)

	crSpecBytes, err := json.Marshal(branchCR)
	if err != nil {
		logger.Error("failed to marshal branch CR spec", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "")
	}
	b.CRSpec = datatypes.JSON(crSpecBytes)

	stepStart = time.Now()
	bizErr = s.createDBRecord(ctx, parentCluster.OrganizationId, b, logger)
	logger.Info("[branch lifecycle] branch create step",
		zap.String("step", "create_db_record"),
		zap.Int64("duration_ms", time.Since(stepStart).Milliseconds()))
	if bizErr != nil {
		return nil, bizErr
	}

	// Send events center
	detail := event.BranchDetail{
		ID:             b.ID,
		Name:           b.Name,
		TenantID:       b.TenantID,
		ProjectID:      xProjectId,
		MainBranchID:   b.MainBranchID,
		ParentBranchID: b.ParentID,
		Annotations:    annotations,
	}
	detailString, err := json.Marshal(detail)
	if err != nil {
		logger.Error("failed to marshal branch detail", zap.Error(err))
	}
	data := &core_event.DisplayEvent{
		DisplayEventType: event.EventTypeCreateBranch,
		DisplayStatus:    core_event.EventStatusStarted,
		Severity:         core_event.EventSeverityInfo,
		Summary:          fmt.Sprintf("Start to create branch %s.", b.Name),
		HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
		AuthorID:         b.CreatedByID,
		ClusterID:        clusterID,
		ProjectID:        xProjectId,
		TenantID:         b.TenantID,
		Detail:           detailString,
	}
	go func(data *core_event.DisplayEvent) {
		if err := s.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: core_event.DisplayEventTopic,
			Data: data,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(data)

	logger.Info("[branch lifecycle] create branch successfully",
		zap.Int64("duration_ms", time.Since(funcStart).Milliseconds()))

	branch := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		Kubernetes:             parentCluster.GetServerlessCluster().GetCluster().GetKubernetes(),
		DisplayName:            b.Name,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentDisplayName:      parentCluster.ClusterName,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		Annotations:            annotations,
		CreatedBy:              b.CreatedBy,
	}
	branch.GenerateBranchName()
	return branch, nil
}

func (s *branchService) createBranchFromBranch(
	ctx context.Context,
	clusterId, branchName, parentBranchID string,
	parentTimestamp *time.Time,
	rootPassword string,
	userID uint64,
	annotations map[string]string,
) (*vo.Branch, bizerr.BizError) {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId), zap.String("branch_name", branchName), zap.String("parent_branch_id", parentBranchID))
	// generate the branch id up front so every log line, including the
	// validation-step timings, carries branch_id
	// k8s resource name must be lowercase. https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#dns-subdomain-names
	id := entity.GenerateIDWithPrefix(constant.BranchIDPrefix)
	logger = logger.With(zap.String("branch_id", id))
	funcStart := time.Now()
	parentBranch, bizErr := s.branchRepo.GetBranchByID(ctx, parentBranchID)
	if bizErr != nil {
		logger.Error("failed to get branch from db", zap.Error(bizErr))
		return nil, bizErr
	}

	if parentBranch.MainBranchID != clusterId {
		logger.Error("branch not belong to main branch")
		return nil, bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
	}

	if parentBranch.Status != constant.BranchStatusRunning {
		logger.Error(
			"only support create branch from running branch",
			zap.String("branch_status", string(parentBranch.Status)),
		)
		return nil, bizerr.New(errorcode.ErrCodeBranchNotRunning, "")
	}

	bizErr = s.exceedNumberLimit(ctx, parentBranch.TenantID)
	if bizErr != nil {
		logger.Error("failed to check exceed number limit", zap.Error(bizErr))
		return nil, bizErr
	}

	clusterIdUint64, err := strconv.ParseUint(clusterId, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	stepStart := time.Now()
	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, clusterIdUint64, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_FULL)
	logger.Info("[branch lifecycle] branch create step",
		zap.String("step", "mgmt_get_cluster"),
		zap.Int64("duration_ms", time.Since(stepStart).Milliseconds()))
	if bizErr != nil {
		logger.Error("failed to get cluster from mgmt svc", zap.Error(bizErr))
		return nil, bizErr
	}
	if cluster.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode] == constant.RuLimitModeThrottled {
		return nil, bizerr.New(errorcode.ErrCodeCreateBranchFromThrottledCluster, "")
	}

	// set annotation
	annotationJson, err := marshalAnnotations(annotations)
	if err != nil {
		logger.Error("failed to marshal annotations", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInvalidBranchAnnotation, "")
	}

	salt, err := entity.GenerateSalt()
	if err != nil {
		logger.Error("failed to generate branch salt", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	hasSetRootPwd := false
	if rootPassword != "" {
		encryptedRootPassword, err := s.encryptorFacade.EncryptBase64(rootPassword)
		if err != nil {
			logger.Error("failed to encrypt root password", zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
		}
		rootPassword = encryptedRootPassword
		hasSetRootPwd = true
	}
	// branches.createdAt is of type TIMESTAMP. It defaults to second precision. If we don't truncate, then
	// branch.MetadataOfBranch will have more precision than the final entity, leading to playground password
	// calculation failed. (2023-04-04T12:46:56.921268434Z vs 2023-04-04T12:46:57Z)
	now := time.Now().Truncate(time.Second)
	if parentTimestamp != nil {
		// parent timestamp need to truncate to seconds precision
		truncateParentTimestamp := parentTimestamp.Truncate(time.Second)
		parentTimestamp = &truncateParentTimestamp
	}
	backupRetention := cluster.GetServerlessCluster().GetBackupPolicy().GetRetentionDays()
	parentTimestamp, err = verifyParentTimestamp(parentTimestamp, parentBranch.CreatedAt, backupRetention, now)
	if err != nil {
		logger.Error("failed to verify parent timestamp", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInvalidParentTimestamp, "%s", err.Error())
	}
	parentTimestamp = parentTimestampOrDefault(
		parentTimestamp,
		config.Global().BizConfig.BranchOmitParentTimestampKuberneteses,
		cluster.GetServerlessCluster().GetCluster().GetKubernetes(),
		now,
	)
	infraProjectId := cluster.GetOriginalProjectId()
	xProjectId := cluster.ProjectId
	b := &entity.Branch{
		ID:                    id,
		CreatedAt:             now,
		UpdatedAt:             &now,
		DeletedTime:           nil,
		CreationFinishedTime:  nil,
		TenantID:              parentBranch.TenantID,
		ProjectID:             infraProjectId,
		CreatedByID:           userID,
		CloudProviderName:     parentBranch.CloudProviderName,
		RegionName:            parentBranch.RegionName,
		Kubernetes:            parentBranch.Kubernetes,
		Name:                  branchName,
		Status:                constant.BranchStatusPending,
		ParentID:              parentBranch.ID,
		ParentTimestamp:       parentTimestamp,
		MainBranchID:          parentBranch.MainBranchID,
		Salt:                  salt,
		Annotations:           annotationJson,
		CreatedBy:             buildCreatedBy(ctx, annotations),
		EncryptedRootPassword: rootPassword,
		HasSetPwd:             hasSetRootPwd,
	}

	parentBranchCR, err := s.infraFacade.GetBranch(ctx, infraProjectId, parentBranch.ID)
	if err != nil {
		logger.Error("failed to get branch CR", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}

	if b.Kubernetes == "" {
		b.Kubernetes = parentBranchCR.Labels[constant.LabelKeyKubernetes]
	}

	branchCR := &cloudv1.Branch{
		ObjectMeta: metav1.ObjectMeta{
			Annotations: copyAnnotations(ctx, parentBranchCR),
			Labels:      copyLabels(ctx, parentBranchCR),
			Namespace:   parentBranchCR.Namespace,
			Name:        id,
		},
		Spec: cloudv1.BranchSpec{
			SqlEndpoints: cloudv1.BranchSqlEndpointsSpec{
				Public: cloudv1.BranchPublicSpec{
					Service: cloudv1.Service{
						Enabled: pointer.Bool(
							!cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().GetDisabled(),
						),
					},
					AuthorizedNetworks: ConvertRegionalToInfraAPIAuthorizedNetworks(
						cluster.GetServerlessCluster().GetCluster().GetEndpoints().GetPublic().GetAuthorizedNetwork(),
					),
				},
				PrivateLinkService: &cloudv1.PrivateLinkService{
					AWS: &cloudv1.AWSPrivateLinkService{AcceptedEndpoints: []string{"*"}},
				},
			},
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{Replicas: 1},
				PD:   cloudv1.PD{Replicas: 1},
				TiKV: cloudv1.TiKV{Replicas: 1},
			},
			AutoScaling: &cloudv1.AutoScaling{
				MinRcu: cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMinRcu(),
				MaxRcu: cluster.GetServerlessCluster().GetCluster().GetAutoScaling().GetMaxRcu(),
			},
			RecoveryMode: false,
		},
		Status: cloudv1.BranchStatus{
			Phase: cloudv1.BranchPhasePending,
		},
	}
	j, err := json.Marshal(b.MetadataOfBranch())
	if err != nil {
		logger.Error("Can't marshal branch metadata", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "")
	}
	branchCR.Annotations[constant.BranchMetadataAnnKey] = string(j)
	branchCR.Annotations[constant.ParentBranchIdAnnKey] = parentBranch.ID
	branchCR.Annotations[constant.MainBranchIdAnnKey] = parentBranch.MainBranchID
	branchCR.Annotations[constant.ParentTypeAnnKey] = constant.ParentTypeBranch
	setParentTimestampAnnotation(branchCR.Annotations, parentTimestamp)

	crSpecBytes, err := json.Marshal(branchCR)
	if err != nil {
		logger.Error("failed to marshal branch CR spec", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "")
	}
	b.CRSpec = datatypes.JSON(crSpecBytes)

	stepStart = time.Now()
	bizErr = s.createDBRecord(ctx, parentBranch.TenantID, b, logger)
	logger.Info("[branch lifecycle] branch create step",
		zap.String("step", "create_db_record"),
		zap.Int64("duration_ms", time.Since(stepStart).Milliseconds()))
	if bizErr != nil {
		return nil, bizErr
	}

	// Send events center
	detail := event.BranchDetail{
		ID:             b.ID,
		Name:           b.Name,
		TenantID:       b.TenantID,
		ProjectID:      xProjectId,
		MainBranchID:   b.MainBranchID,
		ParentBranchID: b.ParentID,
		Annotations:    annotations,
	}
	detailString, err := json.Marshal(detail)
	if err != nil {
		logger.Error("failed to marshal branch detail", zap.Error(err))
	}
	data := &core_event.DisplayEvent{
		DisplayEventType: event.EventTypeCreateBranch,
		DisplayStatus:    core_event.EventStatusStarted,
		Severity:         core_event.EventSeverityInfo,
		Summary:          fmt.Sprintf("Start to create branch %s.", b.Name),
		HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
		AuthorID:         b.CreatedByID,
		ClusterID:        clusterIdUint64,
		ProjectID:        xProjectId,
		TenantID:         b.TenantID,
		Detail:           detailString,
	}
	go func(data *core_event.DisplayEvent) {
		if err := s.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: core_event.DisplayEventTopic,
			Data: data,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(data)

	logger.Info("[branch lifecycle] create branch successfully",
		zap.Int64("duration_ms", time.Since(funcStart).Milliseconds()))
	branch := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		Kubernetes:             branchCR.Labels[constant.LabelKeyKubernetes],
		DisplayName:            b.Name,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentDisplayName:      parentBranch.Name,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		Annotations:            annotations,
		CreatedBy:              b.CreatedBy,
	}
	branch.GenerateBranchName()
	return branch, nil
}

func (s *branchService) createDBRecord(
	ctx context.Context,
	tenantID uint64,
	b *entity.Branch,
	logger *zap.Logger,
) bizerr.BizError {
	err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		lock := &entity.Lock{
			Key:         fmt.Sprintf("create-branch-org-%d", tenantID),
			TimeoutSecs: 10,
		}
		err := s.lockRepo.Lock(ctx, lock)
		if err != nil {
			logger.Error("failed to get lock", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeDB, "")
		}
		logger.Info("get lock successfully")

		defer func() {
			rctx := context.WithoutCancel(ctx)
			err := s.lockRepo.Release(rctx, lock)
			if err != nil {
				logger.Error("failed to release lock", zap.Error(err))
			} else {
				logger.Info("release lock successfully")
			}
		}()

		bizErr := s.exceedNumberLimit(ctx, tenantID)
		if bizErr != nil {
			logger.Error("failed to check exceed number limit", zap.Error(bizErr))
			return bizErr
		}

		bizErr = s.branchRepo.Create(ctx, b)
		if bizErr != nil {
			logger.Error("failed to create branch record", zap.Error(bizErr))
			return bizErr
		}

		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return e
		}

		logger.Error("failed to create branch record", zap.Error(err))
		return bizerr.New(bizerr.ErrCodeDB, "")
	}

	return nil
}

func (s *branchService) exceedNumberLimit(ctx context.Context, tenantId uint64) bizerr.BizError {
	count, bizErr := s.branchRepo.CountBranchesByOrgID(ctx, tenantId)
	if bizErr != nil {
		return bizErr
	}
	var quota int64
	quota, ok := config.Global().BizConfig.BranchTenantNumberQuota[tenantId]
	if !ok {
		quota = config.Global().BizConfig.MaxBranchNumber
	}
	if count >= quota {
		return bizerr.New(errorcode.ErrCodeGreaterThanBranchNumbersLimit,
			"This organization can create up to %d branches across all clusters.", quota)
	}
	return nil
}

func copyLabels(ctx context.Context, parent any) map[string]string {
	labels := make(map[string]string)
	switch parent := parent.(type) {
	case *tidbmgmtservicev1.Cluster:
		labels[constant.LabelKeyProviderType] = parent.GetServerlessCluster().GetCluster().GetProviderType()
		labels[constant.LabelKeyKubernetes] = parent.GetServerlessCluster().GetCluster().GetKubernetes()
		labels[constant.LabelKeyRegion] = parent.Region
		labels[cloudv1.LabelEncryptKeyspace] = strconv.FormatBool(parent.GetServerlessCluster().GetCluster().GetEnhancedEncryptionEnabled())
	case *cloudv1.Branch:
		labels[constant.LabelKeyProviderType] = parent.Labels[constant.LabelKeyProviderType]
		labels[constant.LabelKeyKubernetes] = parent.Labels[constant.LabelKeyKubernetes]
		labels[constant.LabelKeyRegion] = parent.Labels[constant.LabelKeyRegion]
		labels[cloudv1.LabelEncryptKeyspace] = parent.Labels[cloudv1.LabelEncryptKeyspace]
	default:
		log.WithContextL(ctx).Error("unknown parent type", zap.Any("parent", parent))
	}
	return labels
}

func copyAnnotations(ctx context.Context, parent any) map[string]string {
	annotations := make(map[string]string)
	switch parent := parent.(type) {
	case *tidbmgmtservicev1.Cluster:
		annotations[constant.AnnoTidbPoolSelector] = parent.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoTidbPoolSelector]
		annotations[constant.AnnoRuLimitMode] = parent.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoRuLimitMode]
		annotations[constant.AnnoServicePlan] = parent.GetServerlessCluster().GetCluster().GetAnnotations()[constant.AnnoServicePlan]
	case *cloudv1.Branch:
		annotations[constant.AnnoTidbPoolSelector] = parent.Annotations[constant.AnnoTidbPoolSelector]
		annotations[constant.AnnoRuLimitMode] = parent.Annotations[constant.AnnoRuLimitMode]
		annotations[constant.AnnoServicePlan] = parent.Annotations[constant.AnnoServicePlan]
	default:
		log.WithContextL(ctx).Error("unknown parent type", zap.Any("parent", parent))
	}
	return annotations
}

func (s *branchService) ListBranchesInPage(
	ctx context.Context,
	mainBranchID string,
	opt vo.ListBranchesOpt,
) (_ []*vo.Branch, _ int64, bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("ListBranchesInPage", bizErr)
	}()
	logger := log.WithContextL(ctx).With(zap.String("main_branch_id", mainBranchID))

	if opt.Page == 0 {
		opt.Page = 1
	}
	if opt.PageSize == 0 {
		opt.PageSize = 10
	}

	bs, bizErr := s.branchRepo.ListBranchesByMainBranchIDInPage(ctx, mainBranchID, opt)
	if bizErr != nil {
		logger.Error("failed to list branches", zap.Error(bizErr))
		return nil, 0, bizErr
	}
	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, 0, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, 0, bizErr
	}

	branches := make([]*vo.Branch, 0, len(bs))
	for _, b := range bs {
		bv := &vo.Branch{
			ID:                   b.ID,
			CreatedAt:            b.CreatedAt,
			CreationFinishedTime: b.CreationFinishedTime,
			TenantID:             b.TenantID,
			CreatedByID:          b.CreatedByID,
			CloudProviderName:    b.CloudProviderName,
			RegionName:           b.RegionName,
			DisplayName:          b.Name,
			Status:               b.Status,
			ParentID:             b.ParentID,
			ParentTimestamp:      b.ParentTimestamp,
			MainBranchID:         b.MainBranchID,
			UpdateAt:             b.UpdatedAt,
			DeleteAt:             b.DeletedTime,
			HasSetPwd:            b.HasSetPwd,
			CreatedBy:            b.CreatedBy,
			InfraProjectId:       b.ProjectID,
		}
		bv.GenerateBranchName()
		if opt.FullBranch {
			err := s.fullBranchWithBranchCR(ctx, b, bv)
			if err != nil {
				logger.Error("failed to full branch with branch CR", zap.Error(err))
				return nil, 0, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
			}
		}
		// set annotation
		annotationsMap, err := b.UnmarshalAnnotations()
		if err != nil {
			logger.Error("failed to unmarshal annotation", zap.Error(err))
			return nil, 0, bizerr.New(errorcode.ErrCodeInternalError, "")
		}
		bv.Annotations = annotationsMap
		branches = append(branches, bv)
	}
	bizErr = s.fillParentDisplayName(ctx, parentCluster, branches...)
	if bizErr != nil {
		logger.Error("failed to fill parent display name", zap.Error(bizErr))
		return nil, 0, bizErr
	}

	count, bizErr := s.branchRepo.CountBranchesByMainBranchID(ctx, mainBranchID)
	if bizErr != nil {
		logger.Error("failed to count branches", zap.Error(bizErr))
		return nil, 0, bizErr
	}

	return branches, count, nil
}

func (s *branchService) ListBranches(
	ctx context.Context,
	mainBranchID string,
	deleteAfterTime *time.Time,
) (_ []*vo.Branch, bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("ListBranches", bizErr)
	}()
	logger := log.WithContextL(ctx).With(zap.String("main_branch_id", mainBranchID))

	bs, bizErr := s.branchRepo.ListBranchesByMainBranchID(ctx, mainBranchID, deleteAfterTime)
	if bizErr != nil {
		logger.Error("failed to list branches", zap.Error(bizErr))
		return nil, bizErr
	}

	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, bizErr
	}
	branches := make([]*vo.Branch, 0, len(bs))
	for _, b := range bs {
		bv := &vo.Branch{
			ID:                   b.ID,
			CreatedAt:            b.CreatedAt,
			CreationFinishedTime: b.CreationFinishedTime,
			TenantID:             b.TenantID,
			CreatedByID:          b.CreatedByID,
			CloudProviderName:    b.CloudProviderName,
			RegionName:           b.RegionName,
			DisplayName:          b.Name,
			Status:               b.Status,
			ParentID:             b.ParentID,
			ParentTimestamp:      b.ParentTimestamp,
			MainBranchID:         b.MainBranchID,
			CreatedBy:            b.CreatedBy,
			InfraProjectId:       b.ProjectID,
		}
		bv.GenerateBranchName()
		// set annotation
		annotationsMap, err := b.UnmarshalAnnotations()
		if err != nil {
			logger.Error("failed to unmarshal annotation", zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
		}
		bv.Annotations = annotationsMap
		branches = append(branches, bv)
	}
	bizErr = s.fillParentDisplayName(ctx, parentCluster, branches...)
	if bizErr != nil {
		logger.Error("failed to fill parent display name", zap.Error(bizErr))
		return nil, bizErr
	}

	return branches, nil
}

func (s *branchService) GetBranchById(
	ctx context.Context,
	mainBranchID string,
	branchId string,
) (_ *vo.Branch, bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("GetBranchById", bizErr)
	}()
	logger := log.WithContextL(ctx).With(zap.String("main_branch_id", mainBranchID), zap.String("branch_id", branchId))

	b, bizErr := s.branchRepo.GetBranchByID(ctx, branchId)
	if bizErr != nil {
		logger.Error("failed to get branch", zap.Error(bizErr))
		return nil, bizErr
	}

	if b.MainBranchID != mainBranchID {
		logger.Error("branch not belong to main branch")
		return nil, bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
	}

	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, bizErr
	}

	branch := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		DisplayName:            b.Name,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		CreatedBy:              b.CreatedBy,
		InfraProjectId:         b.ProjectID,
	}
	branch.GenerateBranchName()
	err1 := s.fullBranchWithBranchCR(ctx, b, branch)
	if err1 != nil {
		logger.Error("failed to full branch with branch CR", zap.Error(err1))
		return nil, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}
	bizErr = s.fillParentDisplayName(ctx, parentCluster, branch)
	if bizErr != nil {
		logger.Error("failed to fill parent display name", zap.Error(bizErr))
		return nil, bizErr
	}
	// set annotation
	annotationsMap, err := b.UnmarshalAnnotations()
	if err != nil {
		logger.Error("failed to unmarshal annotation", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	branch.Annotations = annotationsMap
	return branch, nil
}

func (s *branchService) DeleteBranchById(
	ctx context.Context,
	mainBranchID string,
	branchId string,
	userID uint64,
) (_ *vo.Branch, bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("DeleteBranchById", bizErr)
	}()
	logger := log.WithContextL(ctx).With(zap.String("main_branch_id", mainBranchID), zap.String("branch_id", branchId))
	funcStart := time.Now()

	b, bizErr := s.branchRepo.GetBranchByID(ctx, branchId)
	if bizErr != nil {
		logger.Error("failed to get branch", zap.Error(bizErr))
		return nil, bizErr
	}

	if b.MainBranchID != mainBranchID {
		logger.Error("branch not belong to main branch")
		return nil, bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
	}

	children, bizErr := s.branchRepo.ListBranchesByParentID(ctx, mainBranchID, branchId)
	if bizErr != nil {
		logger.Error("failed to get child branches", zap.Error(bizErr))
		return nil, bizErr
	}
	if len(children) > 0 {
		return nil, bizerr.New(errorcode.ErrCodeDeleteBranchHasChildren, "")
	}

	err := s.infraFacade.DeleteBranch(ctx, b.ProjectID, b.ID)
	if err != nil {
		logger.Error("failed to delete branch CR", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
	}

	now := time.Now()
	b.Status = constant.BranchStatusDeleted
	b.DeletedTime = &now
	b.UpdatedAt = &now
	bizErr = s.branchRepo.Update(ctx, b, []string{"status", "deleted_time", "updated_at"}, nil)
	if bizErr != nil {
		logger.Error("failed to delete branch after CR deleted", zap.Error(bizErr))
		return nil, bizErr
	}

	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, bizErr
	}

	go s.sendBranchDeleteDisplayEvent(ctx, parentCluster, b, userID)
	logger.Info("[branch lifecycle] delete branch successfully",
		zap.String("branch_name", b.Name),
		zap.Int64("duration_ms", time.Since(funcStart).Milliseconds()))

	annotationsMap, err := b.UnmarshalAnnotations()
	if err != nil {
		logger.Error("failed to unmarshal annotation", zap.Any("annotations", b.Annotations), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	branch := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		DisplayName:            b.Name,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		Annotations:            annotationsMap,
		CreatedBy:              b.CreatedBy,
	}
	branch.GenerateBranchName()
	err = s.fillParentDisplayName(ctx, parentCluster, branch)
	if err != nil {
		// don't return, ignore the error
		logger.Error("failed to fill parent display name", zap.Error(err))
	}

	return branch, nil
}

func (s *branchService) DeleteBranchByClusterId(
	ctx context.Context,
	clusterId string,
	userID uint64,
) (bizErr bizerr.BizError) {
	defer func() {
		collectMetrics("DeleteBranchByClusterId", bizErr)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId))

	branches, err := s.branchRepo.ListBranchesByMainBranchID(ctx, clusterId, nil)
	if err != nil {
		logger.Error("list branches by main branch id failed", zap.Error(err))
		return err
	}
	if len(branches) == 0 {
		logger.Info("no child branches")
		return nil
	}

	clusterIdUint64, err1 := strconv.ParseUint(clusterId, 10, 64)
	if err1 != nil {
		return bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetClusterIncludeDeleted(ctx, clusterIdUint64)
	if bizErr != nil {
		return bizErr
	}

	for _, b := range branches {
		branchStart := time.Now()
		err := s.infraFacade.DeleteBranch(ctx, b.ProjectID, b.ID)
		if err != nil {
			logger.Error("failed to delete branch CR", zap.String("branch_id", b.ID), zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}
		now := time.Now()
		b.Status = constant.BranchStatusDeleted
		b.DeletedTime = &now
		b.UpdatedAt = &now
		bizErr = s.branchRepo.Update(ctx, b, []string{"status", "deleted_time", "updated_at"}, nil)
		if bizErr != nil {
			logger.Error("failed to delete branch after CR deleted", zap.String("branch_id", b.ID), zap.Error(bizErr))
			return bizErr
		}

		go s.sendBranchDeleteDisplayEvent(ctx, parentCluster, b, userID)
		logger.Info("[branch lifecycle] delete branch successfully",
			zap.String("branch_id", b.ID),
			zap.String("branch_name", b.Name),
			zap.Int64("duration_ms", time.Since(branchStart).Milliseconds()))
	}

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

func (s *branchService) sendBranchDeleteDisplayEvent(
	ctx context.Context,
	c *tidbmgmtservicev1.Cluster,
	b *entity.Branch,
	userID uint64,
) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", b.ID))
	logger.Info("start to send branch delete display event")
	annotationsMap, err := b.UnmarshalAnnotations()
	if err != nil {
		logger.Error("failed to unmarshal annotation", zap.Any("annotations", b.Annotations), zap.Error(err))
	}
	clusterID, err := strconv.ParseUint(b.MainBranchID, 10, 64)
	if err != nil {
		logger.Error("failed to parse cluster id", zap.String("cluster_id", b.MainBranchID), zap.Error(err))
	}
	detail := event.BranchDetail{
		ID:             b.ID,
		Name:           b.Name,
		TenantID:       b.TenantID,
		ProjectID:      c.ProjectId,
		MainBranchID:   b.MainBranchID,
		ParentBranchID: b.ParentID,
		Annotations:    annotationsMap,
	}
	detailString, err := json.Marshal(detail)
	if err != nil {
		logger.Error("failed to marshal branch detail", zap.Error(err))
	}
	data := &core_event.DisplayEvent{
		DisplayEventType: event.EventTypeDeleteBranch,
		DisplayStatus:    core_event.EventStatusSucceeded,
		Severity:         core_event.EventSeverityInfo,
		Summary:          fmt.Sprintf("Branch %s deleted.", b.Name),
		HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
		AuthorID:         userID,
		ClusterID:        clusterID,
		ProjectID:        c.ProjectId,
		TenantID:         b.TenantID,
		Detail:           detailString,
	}
	if err := s.eventbusService.Publish(&tieventbus.Event{
		ID:   shortuuid.New(),
		Type: core_event.DisplayEventTopic,
		Data: data,
	}); err != nil {
		logger.Error("failed to publish events center", zap.Error(err))
	}
	logger.Info("succeed to send branch delete display event")
}

func (s *branchService) UpdateBranchHasSetPwd(
	ctx context.Context,
	branchId string,
	hasSetPwd bool,
) (bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branchId))

	updateBranch := &entity.Branch{ID: branchId}
	updateBranch.HasSetPwd = hasSetPwd
	bizErr = s.branchRepo.Update(ctx, updateBranch, []string{"has_set_pwd"}, nil)
	if bizErr != nil {
		logger.Error("failed to update hasSetPwd", zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func (s *branchService) UpdateBranch(
	ctx context.Context,
	mainBranchID string,
	branchId string,
	branch *vo.UpdateBranchReq,
) (*vo.Branch, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("main_branch_id", mainBranchID), zap.String("branch_id", branchId))

	b, bizErr := s.branchRepo.GetBranchByID(ctx, branchId)
	if bizErr != nil {
		logger.Error("failed to get branch", zap.Error(bizErr))
		return nil, bizErr
	}

	if b.MainBranchID != mainBranchID {
		logger.Error("branch not belong to main branch")
		return nil, bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
	}

	updateBranch := &entity.Branch{ID: branchId}
	if branch.DisplayName != "" {
		if len(branch.DisplayName) > 64 {
			logger.Info("invalid branch name length, should be with 1-64, inclusive")
			return nil, bizerr.New(errorcode.ErrCodeBranchNameInvalidLength, "")
		}
		if strings.EqualFold(branch.DisplayName, reservedBranchName) {
			logger.Info("reserved name is not allowed, please try another name")
			return nil, bizerr.New(errorcode.ErrCodeBranchNameReserved, "")
		}
		updateBranch.Name = branch.DisplayName
	}

	bizErr = s.branchRepo.Update(ctx, updateBranch, []string{"name"}, nil)
	if bizErr != nil {
		logger.Error("failed to update branch", zap.Error(bizErr))
		return nil, bizErr
	}

	annotationsMap, err := b.UnmarshalAnnotations()
	if err != nil {
		logger.Error("failed to unmarshal annotation", zap.Any("annotations", b.Annotations), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "")
	}

	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, bizErr
	}

	// Send events center
	detail := event.BranchDetail{
		ID:             b.ID,
		Name:           b.Name,
		TenantID:       b.TenantID,
		ProjectID:      parentCluster.ProjectId,
		MainBranchID:   b.MainBranchID,
		ParentBranchID: b.ParentID,
		Annotations:    annotationsMap,
		NewName:        branch.DisplayName,
	}
	detailString, err := json.Marshal(detail)
	if err != nil {
		logger.Error("failed to marshal branch detail", zap.Error(err))
	}
	data := &core_event.DisplayEvent{
		DisplayEventType: event.EventTypeUpdateBranch,
		DisplayStatus:    core_event.EventStatusSucceeded,
		Severity:         core_event.EventSeverityInfo,
		Summary:          fmt.Sprintf("Successfully updated branch %s.", b.Name),
		HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
		AuthorID:         b.CreatedByID,
		ClusterID:        clusterID,
		ProjectID:        parentCluster.ProjectId,
		TenantID:         b.TenantID,
		Detail:           detailString,
	}
	go func(data *core_event.DisplayEvent) {
		if err := s.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: core_event.DisplayEventTopic,
			Data: data,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(data)

	logger.Info("update branch successfully")

	branchVO := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		DisplayName:            branch.DisplayName,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		Annotations:            annotationsMap,
		CreatedBy:              b.CreatedBy,
	}
	branchVO.GenerateBranchName()
	err = s.fillParentDisplayName(ctx, parentCluster, branchVO)
	if err != nil {
		// don't return, ignore the error
		logger.Error("failed to fill parent display name", zap.Error(err))
	}
	return branchVO, nil
}

func collectMetrics(method string, bizErr bizerr.BizError) {
	if bizErr != nil {
		errCode := strconv.FormatInt(int64(bizErr.ErrCode()), 10)
		metric.BranchInternalErrCounter.WithLabelValues(method, errCode).Inc()
	}
}

func (s *branchService) CollectMetricsFromDB(ctx context.Context) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Info("start collect metrics from db")
	branchStatusCounts, err := s.branchRepo.CountBranchesByStatus(ctx)
	if err != nil {
		logger.Error("failed to count branches by status", zap.Error(err))
		return err
	}
	statusCount := map[constant.BranchStatus]int{
		constant.BranchStatusPending:  0,
		constant.BranchStatusCreating: 0,
		constant.BranchStatusRunning:  0,
		constant.BranchStatusDeleted:  0,
	}
	for _, b := range branchStatusCounts {
		logger.Info("collect branch status count", zap.String("status", string(b.Status)), zap.Int("count", b.Count))
		statusCount[b.Status] = b.Count
	}

	for status, count := range statusCount {
		metric.BranchNumberGauge.WithLabelValues(string(status)).Set(float64(count))
	}
	return nil
}

func (s *branchService) fullBranchWithBranchCR(ctx context.Context, branch *entity.Branch, branchVO *vo.Branch) error {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branchVO.ID))
	if branch.Status == constant.BranchStatusPending {
		logger.Info("branch is in pending status, skip to get branch CR",
			zap.String("status", string(branch.Status)))
		return nil
	}
	// get branch cr
	var branchCR *cloudv1.Branch
	err := retry.OnError(backoff, api_errors.IsNotFound, func() error {
		cr, err := s.infraFacade.GetBranch(ctx, branch.ProjectID, branchVO.ID)
		if err != nil {
			logger.Warn("failed to get branch from infra svc", zap.Error(err))
			return err
		}
		branchCR = cr
		return nil
	})
	if err != nil {
		if !api_errors.IsNotFound(err) {
			logger.Error("failed to get branch from infra svc", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}
		cr, err := s.infraFacade.GetBranchDirectly(ctx, branch.ProjectID, branchVO.ID)
		if err != nil {
			logger.Error("failed to directly get branch from infra svc", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}
		branchCR = cr
	}

	// get endpoint
	var publicEndpoint *vo.PublicEndpoint
	var privateLinkService *vo.PrivateLinkService
	sqlEndpoints := branchCR.Status.SqlEndpoints
	authorizedNetworks := branchCR.Spec.SqlEndpoints.Public.AuthorizedNetworks
	if sqlEndpoints.PublicEndpoint != nil {
		publicEndpoint = &vo.PublicEndpoint{
			Host:               sqlEndpoints.PublicEndpoint.Host,
			Port:               int32(sqlEndpoints.PublicEndpoint.Port),
			AuthorizedNetworks: convertInfraAPIToVOAuthorizedNetworks(authorizedNetworks),
		}
	}
	if sqlEndpoints.PrivateLinkService != nil {
		privateLinkService = &vo.PrivateLinkService{}
		if sqlEndpoints.PrivateLinkService.AWS != nil {
			privateLinkService.Host = sqlEndpoints.PrivateLinkService.AWS.PrivateDNSName
			privateLinkService.Port = *sqlEndpoints.PrivateLinkService.AWS.Port
			privateLinkService.Aws = &vo.AwsPrivateLinkService{
				ServiceName:       sqlEndpoints.PrivateLinkService.AWS.ServiceName,
				AvailabilityZones: sqlEndpoints.PrivateLinkService.AWS.AvailabilityZones,
			}
		}
	}
	// set endpoint if publicEndpoint or privateLinkService is not null
	if publicEndpoint != nil || privateLinkService != nil {
		branchVO.Endpoint = &vo.Endpoint{
			PublicEndpoint:     publicEndpoint,
			PrivateLinkService: privateLinkService,
		}
	}

	// get rcu
	if branchCR.Spec.AutoScaling != nil {
		branchVO.AutoScaling = vo.AutoScaling{
			MaxRcu: branchCR.Spec.AutoScaling.MaxRcu,
			MinRcu: branchCR.Spec.AutoScaling.MinRcu,
		}
	}

	// get userPrefix
	stats, err := branchCR.AwsFreeTierClusterStatusFromBranch()
	if err != nil {
		logger.Error("failed to get status from branch CR", zap.Error(err))
		return err
	}
	branchVO.UserPrefix = stats.KeyspaceToken
	branchVO.KeyspaceId = stats.KeyspaceId
	branchVO.Kubernetes = branchCR.Labels[constant.LabelKeyKubernetes]
	return nil
}

// ListBranchesRowStorage list branches TiKV storage under one cluster(same provider/region/k8s). Don't use this function in billing.
func (s *branchService) ListBranchesRowStorage(ctx context.Context, branches []*vo.Branch) (map[string]float64, error) {
	// only list running branches, since we only care about the storage usage of running branches.
	runningBranch := make([]*vo.Branch, 0, len(branches))
	for _, b := range branches {
		if b.Status == constant.BranchStatusRunning {
			runningBranch = append(runningBranch, b)
		}
	}
	if len(runningBranch) == 0 {
		return nil, nil
	}
	branches = runningBranch
	provider, region, kubernetes := branches[0].CloudProviderName, branches[0].RegionName, branches[0].Kubernetes
	logger := log.WithContextL(ctx).
		With(zap.String("provider", provider), zap.String("region", region), zap.String("kubernetes", kubernetes))

	keyspaceIds := make([]uint32, len(branches))
	for i, b := range branches {
		keyspaceIds[i] = b.KeyspaceId
	}
	o11yAddr := config.GetO11yUrl(provider, region, kubernetes)
	if o11yAddr == "" {
		logger.Error("can't get o11y regional url")
		return nil, errors.New("can't get o11y regional url")
	}
	branchUsages, err := s.o11yFacade.GetBranchesSkuUsagesByRegion(
		ctx,
		keyspaceIds,
		constant.MetricTypeTikvStorageWithoutQuantile,
		o11yAddr,
	)
	if err != nil {
		logger.Error(
			"get branch sku usage failed.",
			zap.Error(err),
			zap.Uint32s("keyspace_ids", keyspaceIds),
			zap.String("addr", o11yAddr),
		)
		return nil, err
	}

	return branchUsages, nil
}

func marshalAnnotations(annotations map[string]string) (datatypes.JSON, error) {
	if len(annotations) > 0 {
		annotationJson, err := json.Marshal(annotations)
		if err != nil {
			return nil, err
		}
		return annotationJson, nil
	}
	return nil, nil
}

func buildCreatedBy(ctx context.Context, annotations map[string]string) string {
	// rpc with created-by annotation
	if annotations[string(vo.CreatedBy)] != "" {
		return annotations[string(vo.CreatedBy)]
	}
	return utils.BuildCreatorID(ctx)
}

func NewBranchMeta(branch *cloudv1.Branch) (*entity.BranchMetadata, error) {
	v := branch.Annotations[constant.BranchMetadataAnnKey]
	if v == "" {
		return nil, errors.New("meta annotation not found")
	}

	metadata := &entity.BranchMetadata{}
	if err := json.Unmarshal([]byte(v), metadata); err != nil {
		return nil, err
	}

	return metadata, nil
}

func (s *branchService) fillParentDisplayName(
	ctx context.Context,
	main *tidbmgmtservicev1.Cluster,
	bs ...*vo.Branch,
) bizerr.BizError {
	parentBranchIDs := sets.New[string]()
	mainClusterName := main.GetClusterName()
	mainBranchID := strconv.FormatUint(main.GetClusterId(), 10)
	for _, b := range bs {
		if utils.IsBranch(b.ParentID) {
			parentBranchIDs.Insert(b.ParentID)
		}
	}
	parentBranchNames := map[string]string{}
	// get parent branches in batch
	ps, bizErr := s.branchRepo.ListBranchesByMainBranchIDAndIDs(ctx, mainBranchID, parentBranchIDs.UnsortedList())
	if bizErr != nil {
		return bizErr
	}
	// make it a id -> name map for later use
	for _, p := range ps {
		parentBranchNames[p.ID] = p.Name
	}
	// fill parent display name
	for _, b := range bs {
		if b.ParentID == mainBranchID {
			b.ParentDisplayName = mainClusterName
			continue
		}
		b.ParentDisplayName = parentBranchNames[b.ParentID]
		if b.ParentDisplayName == "" {
			return bizerr.New(errorcode.ErrCodeInternalError, "parent branch name not found")
		}
	}
	return nil
}

// parentTimestampOrDefault keeps the legacy default (now) for kubernetes
// clusters not opted into online snapshot restores; opted-in clusters omit
// the timestamp so the downstream manager picks the restore mode.
func parentTimestampOrDefault(
	parentTimestamp *time.Time,
	allowlist []string,
	kubernetes string,
	now time.Time,
) *time.Time {
	if parentTimestamp != nil {
		return parentTimestamp
	}
	if slices.Contains(allowlist, kubernetes) {
		return nil
	}
	return &now
}

func verifyParentTimestamp(
	parentTimestamp *time.Time,
	parentCreatedAt time.Time,
	backupRetention uint64,
	end time.Time,
) (*time.Time, error) {
	if parentTimestamp == nil {
		// No explicit point in time means "latest": keep nil so the CR
		// annotation is omitted and the downstream manager picks the restore
		// mode (online snapshot, or PiTR with its own fallback).
		return nil, nil
	}
	if parentTimestamp.After(end) {
		return nil, errors.New("Parent timestamp is invalid to be later than current time.")
	}
	if parentTimestamp.Before(end.Add(-time.Duration(backupRetention) * 24 * time.Hour)) {
		return nil, fmt.Errorf(
			"Data history only preserve for %d day(s). Earlier parent timestamp is not allowed.",
			backupRetention,
		)
	}
	// Because of the limitation of the TiDB PITR, the parent timestamp should be after the parent create time.
	if parentTimestamp.Before(parentCreatedAt) {
		return nil, errors.New("Parent timestamp is not allowed to be earlier than parent creation time.")
	}

	return parentTimestamp, nil
}

func setParentTimestampAnnotation(annotations map[string]string, parentTimestamp *time.Time) {
	if parentTimestamp == nil {
		return
	}
	annotations[constant.ParentTimestampAnnKey] = parentTimestamp.Format(time.RFC3339)
}

func (s *branchService) ResetBranchById(
	ctx context.Context,
	mainBranchID string,
	branchId string,
) (_ *vo.Branch, bizErr bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branchId))
	funcStart := time.Now()
	var b *entity.Branch
	err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		var bizErr bizerr.BizError
		b, bizErr = s.branchRepo.GetBranchByID(ctx, branchId)
		if bizErr != nil {
			logger.Error("failed to get branch", zap.Error(bizErr))
			return bizErr
		}

		if b.MainBranchID != mainBranchID {
			logger.Error("branch not belong to main branch")
			return bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
		}

		if b.Status != constant.BranchStatusRunning {
			logger.Error("branch is not running")
			return bizerr.New(errorcode.ErrCodeBranchNotRunning, "")
		}

		if ok, bizErr := s.checkParentAndChildBranchesStatus(ctx, b); bizErr != nil {
			logger.Error("failed to check parent and child branches status", zap.Error(bizErr))
			return bizErr
		} else if !ok {
			logger.Error("parent or child branches are not running")
			return bizerr.New(errorcode.ErrCodeBranchParentOrChildNotRunning, "")
		}

		branchCR, err := s.infraFacade.GetBranch(ctx, b.ProjectID, b.ID)
		if err != nil {
			logger.Error("failed to get branch CR", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}

		// reset parent timestamp to now
		now := time.Now()
		parentTimestamp := now
		setParentTimestampAnnotation(branchCR.Annotations, &parentTimestamp)
		branchCR.Annotations[constant.AnnoKeyKeyspaceRestore] = constant.AnnoKeyspaceRestoreStart
		// TODO remove it later
		updatedBranch := branchCR.DeepCopy()

		err = s.infraFacade.UpdateBranch(ctx, updatedBranch)
		if err != nil {
			logger.Error("failed to reset parent timestamp", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}

		branchCR.Status.Phase = cloudv1.BranchPhaseRestoring
		branchCR.Status.Reason = "start to restore"
		err = s.infraFacade.UpdateBranchStatus(ctx, branchCR)
		if err != nil {
			logger.Error("failed to update branchCR status", zap.Error(err))
			return bizerr.New(bizerr.ErrCodeSvcInfraAPI, "")
		}

		// update branch record
		b, bizErr = s.updateRestoreRelatedParams(ctx, b, parentTimestamp)
		if bizErr != nil {
			logger.Error("failed to update branch record", zap.Error(bizErr))
			return bizErr
		}

		return nil
	})
	if err != nil {
		var e bizerr.BizError
		if errors.As(err, &e) {
			return nil, e
		}

		logger.Error("failed to update branch record when reseting", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeDB, "")
	}

	annotationsMap, err := b.UnmarshalAnnotations()
	if err != nil {
		logger.Error("failed to unmarshal annotation", zap.Any("annotations", b.Annotations), zap.Error(err))
	}

	clusterID, err := strconv.ParseUint(mainBranchID, 10, 64)
	if err != nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcBranch, "cluster id is invalid")
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(
		ctx,
		clusterID,
		tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_LOCAL,
	)
	if bizErr != nil {
		return nil, bizErr
	}

	// Send events center
	detail := event.BranchDetail{
		ID:             b.ID,
		Name:           b.Name,
		TenantID:       b.TenantID,
		ProjectID:      parentCluster.ProjectId,
		MainBranchID:   b.MainBranchID,
		ParentBranchID: b.ParentID,
		Annotations:    annotationsMap,
	}
	detailString, err := json.Marshal(detail)
	if err != nil {
		logger.Error("failed to marshal branch detail", zap.Error(err))
	}
	data := &core_event.DisplayEvent{
		DisplayEventType: event.EventTypeResetBranch,
		DisplayStatus:    core_event.EventStatusStarted,
		Severity:         core_event.EventSeverityInfo,
		Summary:          fmt.Sprintf("Start to reset branch %s.", b.Name),
		HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
		AuthorID:         b.CreatedByID,
		ClusterID:        clusterID,
		ProjectID:        parentCluster.ProjectId,
		TenantID:         b.TenantID,
		Detail:           detailString,
	}
	go func(data *core_event.DisplayEvent) {
		if err := s.eventbusService.Publish(&tieventbus.Event{
			ID:   shortuuid.New(),
			Type: core_event.DisplayEventTopic,
			Data: data,
		}); err != nil {
			logger.Error("failed to publish events center", zap.Error(err))
			return
		}
	}(data)

	logger.Info("[branch lifecycle] reset branch successfully",
		zap.Int64("duration_ms", time.Since(funcStart).Milliseconds()))

	branch := &vo.Branch{
		ID:                     b.ID,
		CreatedAt:              b.CreatedAt,
		CreationFinishedTime:   b.CreationFinishedTime,
		TenantID:               b.TenantID,
		CreatedByID:            b.CreatedByID,
		CloudProviderName:      b.CloudProviderName,
		RegionName:             b.RegionName,
		DisplayName:            b.Name,
		Status:                 b.Status,
		ParentID:               b.ParentID,
		ParentTimestamp:        b.ParentTimestamp,
		MainBranchID:           b.MainBranchID,
		UpdateAt:               b.UpdatedAt,
		DeleteAt:               b.DeletedTime,
		EncryptedCloudAdminPwd: b.EncryptedCloudAdminPwd,
		HasSetPwd:              b.HasSetPwd,
		CreatedBy:              b.CreatedBy,
	}
	bizErr = s.fillParentDisplayName(ctx, parentCluster, branch)
	if bizErr != nil {
		logger.Error("failed to fill parent display name", zap.Error(bizErr))
		return nil, bizErr
	}
	branch.GenerateBranchName()
	return branch, nil
}

func (s *branchService) UpdateSqlUser(
	ctx context.Context,
	clusterId, branchId, username, password string,
) bizerr.BizError {
	logger := log.WithContextL(ctx).
		With(zap.String("cluster_id", clusterId), zap.String("branch_id", branchId), zap.String("username", username))

	branch, err := s.branchRepo.GetBranchByID(ctx, branchId)
	if err != nil {
		logger.Error("get branch failed", zap.Error(err))
		return err
	}

	if branch.MainBranchID != clusterId {
		return bizerr.New(errorcode.ErrCodeBranchNotBelongToMainBranch, "")
	}

	if branch.Status != constant.BranchStatusRunning {
		return bizerr.New(errorcode.ErrCodeBranchNotRunning, "")
	}

	branchCr, err1 := s.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err1 != nil {
		logger.Error("get branch cr failed", zap.Error(err1))
		return bizerr.New(errorcode.ErrCodeSvcInfra, "")
	}

	logger = logger.With(zap.String("region", branch.RegionName))
	cloudAdminPwd, err1 := s.encryptorFacade.DecryptBase64(branch.EncryptedCloudAdminPwd)
	if err1 != nil {
		logger.Error("encryptorFacade decrypt failed", zap.Error(err1))
		return bizerr.New(errorcode.ErrCodeSvcEncryptor, "")
	}
	operator := &clusterproxy.ClusterOperator{
		Username:   constant.CloudAdminUsername,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: cloudAdminPwd,
	}
	proxy, err := s.clusterProxyProviderFacade.GetBranchProxy(ctx, branchCr, operator)
	if err != nil {
		logger.Error("get cluster proxy failed", zap.Error(err))
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "")
	}

	err = proxy.ChangePassword(ctx, username, password)
	if err != nil {
		logger.Error("change sql user password failed", zap.Error(err))
		if strings.Contains(err.Error(), "Error 1396 (HY000): Operation ALTER USER failed for") {
			return bizerr.New(bizerr.ErrCodeParams, "ALTER USER failed for '%s'", username)
		}
		return bizerr.New(errorcode.ErrCodeSvcClusterProxy, "")
	}

	// update branch has_set_pwd
	stats, err1 := branchCr.AwsFreeTierClusterStatusFromBranch()
	if err1 != nil {
		logger.Error("failed to get status from branch CR", zap.Error(err1))
		return bizerr.New(bizerr.ErrCodeUnknown, "")
	}
	if username == stats.KeyspaceToken+".root" && !branch.HasSetPwd {
		err := s.UpdateBranchHasSetPwd(ctx, branch.ID, true)
		if err != nil {
			logger.Error("update branch hasSetPwd failed", zap.Error(err))
			return err
		}
	}

	return nil
}

func (s *branchService) updateRestoreRelatedParams(
	ctx context.Context,
	branch *entity.Branch,
	parentTimestamp time.Time,
) (*entity.Branch, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	now := time.Now()
	branch.UpdatedAt = &now
	branch.ParentTimestamp = &parentTimestamp
	branch.CloudAdminPwdSynced = false
	branch.EncryptedCloudAdminPwd = ""
	branch.Status = constant.BranchStatusRestoring
	branch.HasSetPwd = false

	err := s.branchRepo.Update(ctx, branch, []string{
		"updated_at",
		"parent_timestamp",
		"cloud_admin_pwd_synced",
		"encrypted_cloud_admin_pwd",
		"status",
		"has_set_pwd",
	}, nil)
	if err != nil {
		logger.Error("set restore related flags failed")
		return nil, err
	}
	logger.Info("set restore related flags succeed")

	return branch, nil
}

func (s *branchService) checkParentAndChildBranchesStatus(
	ctx context.Context,
	branch *entity.Branch,
) (bool, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	parentID := branch.ParentID
	if !utils.IsBranch(parentID) {
		// note: cluster's status is not controlled by this service, it may change after this check
		if ok, bizErr := s.isParentClusterStatusNormal(ctx, branch); bizErr != nil {
			logger.Error("failed to check parent cluster status", zap.Error(bizErr))
			return false, bizErr
		} else if !ok {
			logger.Error("parent cluster status is not normal")
			return false, nil
		}
	} else {
		parentBranch, err := s.branchRepo.GetBranchByID(ctx, parentID)
		if err != nil {
			logger.Error("get parent branch failed", zap.Error(err))
			return false, err
		}
		if parentBranch.Status != constant.BranchStatusRunning {
			logger.Error(fmt.Sprintf("parent branch %s status is not running", parentID))
			return false, nil
		}
	}

	children, err := s.branchRepo.ListBranchesByParentID(ctx, branch.MainBranchID, branch.ID)
	if err != nil {
		logger.Error("list child branches failed", zap.Error(err))
		return false, err
	}
	for _, child := range children {
		if child.Status != constant.BranchStatusRunning {
			logger.Error(fmt.Sprintf("child branch %s status is not running", child.ID))
			return false, nil
		}
	}
	return true, nil
}

func (s *branchService) isParentClusterStatusNormal(
	ctx context.Context,
	branch *entity.Branch,
) (bool, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))
	clusterID, err := strconv.ParseUint(branch.ParentID, 10, 64)
	if err != nil {
		return false, bizerr.New(errorcode.ErrCodeSvcBranch, "invalid cluster id %s", branch.ParentID)
	}
	parentCluster, bizErr := s.mgmtSvc.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		logger.Error("failed to get cluster from core svc", zap.Error(bizErr))
		return false, bizErr
	}

	if parentCluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		logger.Error(
			"only support restore branch from normal cluster",
			zap.Int("status", int(parentCluster.GetServerlessCluster().GetUserStatus())),
		)
		return false, nil
	}
	return true, nil
}

func ConvertRegionalToInfraAPIAuthorizedNetworks(
	peNetworks []*regionalserverlessv1.PublicEndpoint_AuthorizedNetwork,
) []cloudv1.AuthorizedNetwork {
	authorizedNetworks := make([]cloudv1.AuthorizedNetwork, 0, len(peNetworks))
	for _, peNetwork := range peNetworks {
		authorizedNetwork := cloudv1.AuthorizedNetwork{
			StartIpAddress: peNetwork.StartIpAddress,
			EndIpAddress:   peNetwork.EndIpAddress,
			DisplayName:    peNetwork.DisplayName,
		}
		authorizedNetworks = append(authorizedNetworks, authorizedNetwork)
	}

	return authorizedNetworks
}

func convertInfraAPIToVOAuthorizedNetworks(peNetworks []cloudv1.AuthorizedNetwork) []vo.AuthorizedNetwork {
	authorizedNetworks := make([]vo.AuthorizedNetwork, 0, len(peNetworks))

	for _, peNetwork := range peNetworks {
		authorizedNetwork := vo.AuthorizedNetwork{
			StartIpAddress: peNetwork.StartIpAddress,
			EndIpAddress:   peNetwork.EndIpAddress,
			DisplayName:    peNetwork.DisplayName,
		}
		authorizedNetworks = append(authorizedNetworks, authorizedNetwork)
	}

	return authorizedNetworks
}