package impl

//nolint:gci
import (
	"context"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	filter "github.com/pingcap/tidb/pkg/util/table-filter"
	"github.com/samber/lo"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	bucketRepo "github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.buf.build/protocolbuffers/go/linuxgit/tidbcloud/base"
	"go.uber.org/zap"
	"golang.org/x/sync/semaphore"
)

var (
	validClusterStatesForBackup = []dedicatedv1.Cluster_State{
		dedicatedv1.Cluster_AVAILABLE,
	}
)

var _ backup.SnapshotBackupModel = &SnapshotBackupModel{}

type SnapshotBackupModel struct {
	// db
	dbProvider       db.TransProvider
	backupRepo       backupRepo.BackupRepo
	scheduleRepo     backupRepo.BackupScheduleRepo
	restoreRepo      restoreRepo.RestoreRepo
	logBackupRepo    backupRepo.LogBackupRepo
	bucketRepo       bucketRepo.BucketRepo
	backupExportRepo backupRepo.BackupExportRepo

	// sub models
	backupSettingM backup.BackupSettingModel

	// internal helpers
	job         cluster_job.Model
	workflowCli workflow.Workflow
	coreCli     grpcli.CoreGrpcClient
	mgmtCli     grpcli.MgmtSvcGrpcClient
	importCli   grpcli.ImportGrpcClient
	accountCli  grpcli.AccountSvcGrpcClient

	featureGate feature_gate.FeatureGate
}

type preCheckResp struct {
	cluster *mgmt.Cluster
	setting *entity.BackupSetting
}

func NewSnapshotBackupModel(
	trans db.TransProvider,
	job cluster_job.Model,
	coreCli grpcli.CoreGrpcClient,
	mgmtCli grpcli.MgmtSvcGrpcClient,
	importCli grpcli.ImportGrpcClient,
	accountCli grpcli.AccountSvcGrpcClient,
	workflowCli workflow.Workflow,
	bucketRepo bucketRepo.BucketRepo,
	backupRepo backupRepo.BackupRepo,
	scheduleRepo backupRepo.BackupScheduleRepo,
	restoreRepo restoreRepo.RestoreRepo,
	backupExportRepo backupRepo.BackupExportRepo,
	backupSettingM backup.BackupSettingModel,
	featureGate feature_gate.FeatureGate,
) *SnapshotBackupModel {
	return &SnapshotBackupModel{
		dbProvider:       trans,
		bucketRepo:       bucketRepo,
		backupRepo:       backupRepo,
		scheduleRepo:     scheduleRepo,
		restoreRepo:      restoreRepo,
		backupExportRepo: backupExportRepo,

		backupSettingM: backupSettingM,

		job:         job,
		coreCli:     coreCli,
		mgmtCli:     mgmtCli,
		importCli:   importCli,
		accountCli:  accountCli,
		workflowCli: workflowCli,
		featureGate: featureGate,
	}
}

func (b *SnapshotBackupModel) AsyncCreate(ctx context.Context,
	req *domain.AsyncCreateBackupReq,
) (*domain.AsyncCreateBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("SnapshotBackupModel.AsyncCreate").With(zap.Any("cluster id", req.ClusterID))
	if err := req.ValidateReq(logger); err != nil {
		return nil, err
	}

	resp := &domain.AsyncCreateBackupResp{}

	// do pre check
	withLoggerCtx := log.WithLogger(ctx, logger)
	pRes, preCheckErr := b.preCheck(withLoggerCtx, req)
	if preCheckErr != nil {
		return nil, preCheckErr
	}

	if req.DryRun {
		logger.Info("dry run async create snapshot backup success")
		return nil, nil
	}

	txF := func(ctx context.Context) error {
		// 1: create backup entity
		backupEntity, createErr := b.createDBEntities(ctx, logger, pRes, req)
		if createErr != nil {
			return createErr
		}
		resp.BackupID = backupEntity.ID.Uint64()
		resp.BackupScheduleID = backupEntity.BackupScheduleID.Uint64()
		// create db record success, let's leave backup id in logger ctx for tracing
		logger = logger.With(zap.Uint64("backup id", backupEntity.ID.Uint64()))

		// 2: register cluster job to lock the cluster
		if err := b.job.RegisterBackupClusterJob(ctx, req.ClusterID, backupEntity.ID.Uint64()); err != nil {
			return err
		}
		// 3: create backup dag
		if dagID, err := b.createBackupDag(ctx, logger, req, backupEntity); err != nil {
			return err
		} else {
			resp.WorkflowID = dagID
		}
		return nil
	}
	if err := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		logger.Error("create backup failed", zap.Error(err))
		// when meet db or other internal error, we should return an internal to user rather than raw error
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return resp, nil
}

func (b *SnapshotBackupModel) preCheck(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) (*preCheckResp, bizerr.BizError) {
	setting, settingErr := b.checkBackupSetting(ctx, req)
	if settingErr != nil {
		return nil, settingErr
	}
	if err := b.checkDuplicateName(ctx, req); err != nil {
		return nil, err
	}
	if err := b.checkExpectedBackupTS(ctx, req); err != nil {
		return nil, err
	}
	if err := b.checkTableFilters(ctx, req); err != nil {
		return nil, err
	}
	if err := b.checkRunningBackups(ctx, req); err != nil {
		return nil, err
	}
	cluster, clusterErr := b.checkCluster(ctx, req)
	if clusterErr != nil {
		return nil, clusterErr
	}
	if err := b.checkRunningImports(ctx, req, cluster); err != nil {
		return nil, err
	}
	return &preCheckResp{cluster: cluster, setting: setting}, nil
}

func (b *SnapshotBackupModel) checkExpectedBackupTS(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) bizerr.BizError {
	if req.ExpectedBackupTS == "" {
		return nil
	}

	_, err := strconv.ParseInt(req.ExpectedBackupTS, 10, 64)
	if err != nil {
		log.GetLogger(ctx).Warn("parse ExpectedBackupTS to int64 error",
			zap.String("ExpectedBackupTS", req.ExpectedBackupTS), zap.Error(err))
		return bizerr_wrapper.NewInvalidInputError(config.ErrInvalidParam, "ExpectedBackupTS", req.ExpectedBackupTS)
	}
	return nil
}

func (b *SnapshotBackupModel) checkTableFilters(ctx context.Context, req *domain.AsyncCreateBackupReq) bizerr.BizError {
	if len(req.TableFilter) == 0 {
		return nil
	}

	if _, err := filter.Parse(req.TableFilter); err != nil {
		return bizerr_wrapper.NewInvalidInputError("table filter invalid, err: %s", err.Error())
	}
	return nil
}

func (b *SnapshotBackupModel) checkBackupSetting(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) (*entity.BackupSetting, bizerr.BizError) {
	settingResp, settingErr := b.backupSettingM.QueryBackupSetting(ctx, pointer.ToUint64(req.ClusterID))
	if settingErr != nil {
		log.GetLogger(ctx).Error("failed to get backup setting", zap.Error(settingErr))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	setting := settingResp.BackupSetting
	if !setting.IsEligibleForBackup() {
		log.GetLogger(ctx).Error("backup setting not normal or editing_pitr", zap.Any("status", setting.Status))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return setting, nil
}

func (b *SnapshotBackupModel) checkDuplicateName(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) bizerr.BizError {
	if req.BackupType != dataflow.BackupType_MANUAL {
		return nil
	}
	queryF := func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
		return qs.NameEq(req.Name).ClusterIDEq(entity.ID(req.ClusterID)).
			TypeEq(entity.BackupTypeManual).DeletedAtIsNull()
	}
	cnt, err := b.scheduleRepo.Count(ctx, queryF)
	if err != nil {
		log.GetLogger(ctx).Error("failed to count backup schedule", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	if cnt > 0 {
		return bizerr.New(
			config.ErrCodeAsyncCreateBackupPreCheckDuplicateName,
			config.ErrMsgAsyncCreateBackupPreCheckDuplicateName,
			req.Name, req.ClusterID)
	}
	return nil
}

func (b *SnapshotBackupModel) checkRunningBackups(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) bizerr.BizError {
	// need review here? only check this is enough?
	queryF := func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.ClusterIDEq(entity.ID(req.ClusterID)).StatusIn(
			entity.BackupStatusRunning, entity.BackupStatusPending).
			DeletedAtIsNull()
	}
	cnt, err := b.backupRepo.Count(ctx, queryF)
	if err != nil {
		log.GetLogger(ctx).Error("failed to count backup", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	if cnt > 0 {
		return bizerr.New(
			config.ErrCodeAsyncCreateBackupPreCheckRunningBackups,
			config.ErrMsgAsyncCreateBackupPreCheckRunningBackups,
			req.ClusterID)
	}
	return nil
}

func (b *SnapshotBackupModel) checkCluster(
	ctx context.Context, req *domain.AsyncCreateBackupReq,
) (*mgmt.Cluster, bizerr.BizError) {
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if err != nil {
		log.GetLogger(ctx).Error("get cluster error", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	dc := cluster.GetDedicatedCluster()
	// allow delete status is to support final backup
	if !lo.Contains(validClusterStatesForBackup, dc.GetState()) && dc.GetState() != dedicatedv1.Cluster_DELETED {
		return nil, bizerr.New(
			config.ErrCodeAsyncCreateBackupPreCheckClusterNotNormal,
			config.ErrMsgAsyncCreateBackupPreCheckClusterNotNormal, req.ClusterID, dc.GetState())
	}
	if cluster.DeployType != mgmt.DeployType_DEPLOY_TYPE_DEDICATED {
		log.GetLogger(ctx).Error("dev tier cluster can not create backup", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	return cluster, nil
}

func (b *SnapshotBackupModel) checkRunningImports(ctx context.Context,
	req *domain.AsyncCreateBackupReq, cluster *mgmt.Cluster,
) bizerr.BizError {
	listImportResp, err := b.importCli.List(ctx, &dataflow.ListImportReq{
		OrgId:     cluster.GetOrganizationId(),
		ProjectId: cluster.GetProjectId(),
		ClusterId: cluster.GetClusterId(),
		Statuses: []dataflow.ImportGetResp_Status{
			dataflow.ImportGetResp_PREPARING,
			dataflow.ImportGetResp_IMPORTING,
			dataflow.ImportGetResp_CANCELING,
		},
	})
	if err != nil {
		log.GetLogger(ctx).Error("failed to count imports", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, req.ClusterID)
	}
	if listImportResp != nil && len(listImportResp.Imports) > 0 {
		return bizerr.New(
			config.ErrCodeAsyncCreateBackupPreCheckRunningImports,
			config.ErrMsgAsyncCreateBackupPreCheckRunningImports, req.ClusterID)
	}
	return nil
}

//nolint:funlen
func (b *SnapshotBackupModel) createDBEntities(
	ctx context.Context, logger *zap.Logger, pres *preCheckResp, req *domain.AsyncCreateBackupReq) (
	*entity.Backup, bizerr.BizError,
) {
	bs := &entity.BackupSchedule{}
	// for manual backup, we need create backup entity and backup schedule entity
	if req.BackupType == dataflow.BackupType_MANUAL {
		now := time.Now()
		bsInner := &entity.BackupSchedule{
			NextBackupTime: &now,
			Name:           req.Name,
			Description:    req.Description,
			Type:           entity.BackupTypeManual,
			Method:         entity.BackupMethodFull,
			UserID:         entity.ID(req.UserID),
			ClusterID:      entity.ID(req.ClusterID),
			ClusterType:    entity.BackupClusterTypeDedicated,
		}
		if err := b.scheduleRepo.Create(ctx, bsInner); err != nil {
			return nil, err.Chain("failed to create backup schedule cluster: %d", req.ClusterID)
		}
		bs = bsInner
	} else {
		// for auto backup, the backup schedule entity should be created before
		innerBs, err := b.scheduleRepo.Load(ctx, *req.AutoBackupScheduleID)
		if err != nil {
			return nil, err.Chain("failed to load backup schedule cluster: %d", req.ClusterID)
		}
		bs = innerBs
	}

	spID, createErr := b.coreCli.CreateClusterSnapshot(ctx, req.ClusterID)
	if createErr != nil {
		return nil, createErr.Chain("failed to create cluster snapshot cluster: %d", req.ClusterID)
	}

	masterBucket := pres.setting.QueryMasterBucket()
	if masterBucket == nil {
		return nil, createErr.Chain("failed to get master bucket of cluster: %d", req.ClusterID)
	}

	bucketSetting := pres.setting.Bucket
	piTRSetting := pres.setting.PiTR
	cluster := pres.cluster
	backupEntity := &entity.Backup{
		BackupScheduleID:  bs.ID,
		BackupSchedule:    bs,
		ClusterSnapshotID: entity.ID(spID),
		ClusterVersion:    cluster.GetDedicatedCluster().GetVersion(),
		IsVolumeBackup:    pointer.ToBool(false),
		Status:            entity.BackupStatusPending,
		TenantID:          entity.ID(cluster.GetOrganizationId()),
		ProjectID:         entity.ID(cluster.GetProjectId()),
		ClusterID:         entity.ID(cluster.GetClusterId()),
		CleanPolicy:       entity.BackupCleanPolicyRetain,
		BucketID:          entity.ID(masterBucket.BucketId),
		ExpectedBackupTS:  req.ExpectedBackupTS,
		TableFilter:       req.TableFilter,
		IsInternal:        req.IsInternal,

		PiTRSettingSnapshot:   piTRSetting,
		BucketSettingSnapshot: bucketSetting,
		ExtraInfo: &entity.BackupExtra{
			On3LayerArch:        true,
			BackupCRNameVersion: "v1",
		},
	}
	if err := b.backupRepo.Save(ctx, backupEntity); err != nil {
		return nil, err.Chain("failed to create backup for cluster: %d", req.ClusterID)
	}

	if req.BucketID != 0 && req.Prefix != "" {
		// for specify bucketID and prefix backup
		bucket, err := b.bucketRepo.Load(ctx, req.BucketID)
		if err != nil {
			return nil, err.Chain("failed to load request bucket ID: %d", req.BucketID)
		}
		backupEntity.Storage = fmt.Sprintf("%s/%s/%s",
			bucket.URI(), req.Prefix, backupEntity.CRName())
		backupEntity.BucketID = entity.ID(req.BucketID)
		backupEntity.BucketSettingSnapshot.Prefix = pointer.ToStringOrNil(req.Prefix)
	} else {
		// in order to generate backup id
		bucketResp, bucketErr := b.backupSettingM.QueryMasterBucket(ctx, &domain.QueryMasterBucketReqV2{
			ClusterID: pointer.ToUint64(req.ClusterID),
		})
		if bucketErr != nil {
			return nil, bucketErr.Chain("failed to get bucket of cluster: %d", req.ClusterID)
		}
		backupEntity.Storage = fmt.Sprintf("%s/%s/%s",
			*bucketResp.BucketURI, *bucketSetting.Prefix, backupEntity.CRName())
	}
	backupEntity.BackupSchedule = nil
	if err := b.backupRepo.Save(ctx, backupEntity); err != nil {
		return nil, err.Chain("failed to create backup for cluster: %d", req.ClusterID)
	}
	return backupEntity, nil
}

func (b *SnapshotBackupModel) createBackupDag(
	ctx context.Context,
	logger *zap.Logger,
	req *domain.AsyncCreateBackupReq,
	backupEntity *entity.Backup,
) (string, bizerr.BizError) {
	vars := map[string]string{workflow.KeyBackupID: strconv.FormatUint(uint64(backupEntity.ID), 10)}
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(uint64(backupEntity.ID), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeSnapshotBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionSnapshotBackupCreate),
		workflow.ClusterID:         strconv.FormatUint(uint64(backupEntity.ClusterID), 10),
	}
	dagTemplate := workflow.DagCreateSnapshotBackup
	if req.IsWeekly {
		dagTemplate = workflow.DagCreateWeeklySnapshotBackup
	}
	res, err := b.workflowCli.RunDagWithTags(dagTemplate, vars, tags)
	if err != nil {
		logger.Error("create backup dag failed", zap.Error(err))
		return "", bizerr_wrapper.NewInternalErr("failed to create backup dag")
	}
	return res.ID, nil
}

func (b *SnapshotBackupModel) AsyncDelete(ctx context.Context,
	req *domain.AsyncDeleteBackupReq,
) (*domain.AsyncDeleteBackupResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).Named("SnapshotBackupModel.AsyncDelete").With(zap.Any("backup id", req.BackupID))

	backup, err := b.backupRepo.Load(ctx, req.BackupID)
	if err != nil {
		logger.Warn("failed to load backup", zap.Error(err))
		return nil, bizerr_wrapper.NewNotFoundErr(config.ErrMsgNotFound, req.BackupID)
	}
	resp := &domain.AsyncDeleteBackupResp{}

	// handle deleted backup
	if backup.DeletedAt != nil {
		logger.Warn("backup already deleted")
		return resp, nil
	}

	if err := b.preCheckForAsyncDelete(ctx, backup); err != nil {
		return nil, err
	}

	// delete ops
	txF := func(ctx context.Context) error {
		deletedTime := time.Now()
		if err := b.deleteBackupEntity(ctx, deletedTime, backup); err != nil {
			logger.Error("failed deleteBackupEntity", zap.Error(err))
			return err
		}

		if err := b.deleteBackupScheduleEntityForManualBackup(ctx, backup.BackupSchedule); err != nil {
			logger.Error("failed deleteBackupScheduleEntityForManualBackup", zap.Error(err))
			return err
		}

		if workflowID, err := b.createBackupDeleteDag(ctx, logger, backup); err != nil {
			logger.Error("failed to create delete backup dag", zap.Error(err))
			return err
		} else {
			resp.WorkflowID = workflowID
		}
		return nil
	}

	if err := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		return txF(ctx)
	}); err != nil {
		logger.Error("delete backup failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, backup.ClusterID)
	}
	return resp, nil
}

func (b *SnapshotBackupModel) deleteBackupEntity(
	ctx context.Context, deletedTime time.Time, backup *entity.Backup,
) bizerr.BizError {
	if err := backup.DeleteBackup(deletedTime); err != nil {
		return err
	}
	if err := b.backupRepo.Save(ctx, backup); err != nil {
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, backup.ClusterID)
	}
	return nil
}

// for manual backup, we need mark this schedule as deleted, otherwise when user create a backup with same name
// it will fail with duplicate name error
func (b *SnapshotBackupModel) deleteBackupScheduleEntityForManualBackup(
	ctx context.Context, backupSchedule *entity.BackupSchedule,
) bizerr.BizError {
	if backupSchedule.Type == entity.BackupTypeManual {
		if err := b.scheduleRepo.SoftDelete(ctx, backupSchedule.ID); err != nil {
			return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, backupSchedule.ClusterID)
		}
	}
	return nil
}

func (b *SnapshotBackupModel) createBackupDeleteDag(
	ctx context.Context, logger *zap.Logger, backup *entity.Backup,
) (string, bizerr.BizError) {
	vars := map[string]string{workflow.KeyBackupID: strconv.FormatUint(uint64(backup.ID), 10)}
	tags := map[string]string{
		workflow.BusinessID:        strconv.FormatUint(uint64(backup.ID), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeSnapshotBackup),
		workflow.BusinessActionKey: string(workflow.BusinessActionSnapshotBackupDelete),
		workflow.ClusterID:         strconv.FormatUint(uint64(backup.ClusterID), 10),
	}
	res, err := b.workflowCli.RunDagWithTags(workflow.DagDeleteSnapshotBackup, vars, tags)
	if err != nil {
		return "", bizerr_wrapper.NewInternalErr("failed to create delete backup dag")
	}
	return res.ID, nil
}

func (b *SnapshotBackupModel) Query(
	ctx context.Context, req *domain.QueryBackupReq,
) (*domain.QueryBackupResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backup, err := b.backupRepo.Get(ctx,
		b.preloadBackupScheduleOp(),
		func(qs entity.BackupQuerySet) entity.BackupQuerySet {
			if len(req.Filter.BackupIDs) > 0 {
				qs = qs.IDIn(pie.Map(req.Filter.BackupIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
			}
			return qs
		})
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr_wrapper.NewNotFoundErr(config.ErrMsgRestoreNotFound)
		}
		logger.Error("query restore failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}
	return &domain.QueryBackupResp{
		BackupEntity: backup,
		BackupID:     uint64(backup.ID),
		ClusterID:    uint64(backup.ClusterID),
	}, nil
}

func (b *SnapshotBackupModel) List(
	ctx context.Context, req *domain.ListBackupReq,
) (*domain.ListBackupResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	backups, err := b.backupRepo.ListExternal(ctx,
		b.preloadBackupScheduleOp(),
		b.listBackupsFilterOp(req.Filter),
		b.listBackupsPagerOp(req.Pager),
		b.orderByBackupCreateAtDescOp(),
	)
	if err != nil {
		logger.Error("list backup failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}
	count, err := b.backupRepo.CountExternal(ctx,
		b.listBackupsFilterOp(req.Filter),
	)
	if err != nil {
		return nil, err
	}
	if len(backups) == 0 {
		return &domain.ListBackupResp{Total: count}, nil
	}

	items, err := b.convertListBackupResp(ctx, backups)
	if err != nil {
		return nil, err
	}
	return &domain.ListBackupResp{
		Items: items,
		Total: count,
	}, nil
}

func (b *SnapshotBackupModel) convertListBackupResp(
	ctx context.Context, backups []entity.Backup,
) ([]domain.ListBackupRespItem, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	clusterSnapshotMap, err := b.queryClusterSnapshotsForBackups(ctx, backups)
	if err != nil {
		logger.Error("query cluster snapshots for backups failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}

	cloudProviderNames := pie.Map(pie.Values(clusterSnapshotMap), func(s *core.ClusterSnapshot) string {
		return s.CloudProviderName
	})
	cloudRegions, err := b.coreCli.ListCloudRegions(ctx, &core.ListCloudRegionsReq{
		Filter: &core.CloudRegionFilter{
			CloudProviderNames: cloudProviderNames,
		},
	})
	if err != nil {
		logger.Error("list cloud regions failed", zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}
	if m, ok := cloudRegions[cloudv1.CloudProviderAWS]; ok {
		if _, ok = m[config.AWSRegionNameOsaka]; !ok {
			m[config.AWSRegionNameOsaka] = &core.CloudRegion{
				Name:              config.AWSRegionNameOsaka,
				CloudProviderName: cloudv1.CloudProviderAWS,
				DisplayName:       config.AWSRegionDisplayNameOsaka,
			}
		}
	}

	items := pie.Map(backups, func(backup entity.Backup) domain.ListBackupRespItem {
		snapshot := clusterSnapshotMap[uint64(backup.ClusterSnapshotID)]
		if snapshot == nil {
			logger.Warn("cluster snapshot not found", zap.Any("backup", backup))
			return b.convertListBackupItemResp(ctx, backup, nil, nil)
		} else {
			return b.convertListBackupItemResp(ctx, backup, snapshot, cloudRegions[snapshot.CloudProviderName])
		}
	})
	return items, nil
}

func (b *SnapshotBackupModel) convertDataflowRegion(
	region string, regionMap map[string]*core.CloudRegion,
) *domain.RegionInfo {
	regionInfo := &domain.RegionInfo{
		Name:        region,
		DisplayName: region,
	}
	if region, ok := regionMap[region]; ok {
		regionInfo.DisplayName = region.DisplayName
	}
	return regionInfo
}

func (b *SnapshotBackupModel) convertListBackupItemResp(
	ctx context.Context, backup entity.Backup, clusterSnapshot *core.ClusterSnapshot,
	cloudRegions map[string]*core.CloudRegion,
) domain.ListBackupRespItem {
	item := domain.ListBackupRespItem{
		ID:             uint64(backup.ID),
		ClusterID:      uint64(backup.ClusterID),
		Name:           backup.Name(),
		ClusterVersion: backup.ClusterVersion,
		Description:    backup.BackupSchedule.Description,
		CreatedTime:    backup.CreatedAt.UTC(),
		Size:           uint64(backup.BackupSize),
		Status:         domain.BackupStatusEntity2Domain(backup.Status),
		Type:           domain.BackupTypeEntity2Domain(backup.BackupSchedule.Type),
	}
	if len(backup.TableFilter) != 0 {
		item.TableFilter = append(item.TableFilter, backup.TableFilter...)
	}
	clusterConfig := b.convertClusterComponents(clusterSnapshot)
	if clusterConfig != nil {
		item.ClusterConfig = *clusterConfig
	}
	if backup.BucketSettingSnapshot != nil && len(cloudRegions) != 0 {
		pie.Of(backup.BucketSettingSnapshot.Buckets).Each(func(bucket entity.BackupSettingBucket) {
			for _, region := range bucket.Regions {
				item.Regions = append(item.Regions, *b.convertDataflowRegion(region, cloudRegions))
			}
		})
	}

	item.ExpirationTime = b.calculateBackupExpirationTime(backup)
	item.BackupTs = backup.BackupTS(ctx, util.ParseTSFromTSString)
	return item
}

func (b *SnapshotBackupModel) calculateBackupExpirationTime(backup entity.Backup) *time.Time {
	if backup.BackupSchedule.Type != entity.BackupTypeAuto {
		return nil
	}
	period := backup.BackupSchedule.GetAvailableRetentionPeriod()
	expireDuration := time.Duration(period) * config.TimeDay
	return pointer.To(backup.CreatedAt.UTC().Add(expireDuration))
}

func (b *SnapshotBackupModel) convertClusterComponents(
	clusterSnapshot *core.ClusterSnapshot,
) *domain.ClusterConfig {
	if clusterSnapshot == nil || len(clusterSnapshot.Components) == 0 {
		return nil
	}

	clusterConfig := &domain.ClusterConfig{
		Components: &domain.ClusterComponents{},
	}
	pie.Of(clusterSnapshot.Components).Each(func(component *core.ClusterComponent) {
		if component.ComponentType == core.ComponentType_COMPONENT_TYPE_TIPROXY {
			clusterConfig.Components.TiProxy = &domain.TiProxyComponent{
				Compute: domain.Compute{
					CpuNum:    component.CpuNum,
					MemoryGib: component.MemoryGi,
				},
				NodeSize:     component.NodeProfileName,
				NodeQuantity: component.Replicas,
			}
		}
		if component.ComponentType == core.ComponentType_COMPONENT_TYPE_TIDB {
			clusterConfig.Components.Tidb = &domain.TiDBComponent{
				Compute: domain.Compute{
					CpuNum:    component.CpuNum,
					MemoryGib: component.MemoryGi,
				},
				NodeSize:     component.NodeProfileName,
				NodeQuantity: component.Replicas,
			}
		}
		if component.ComponentType == core.ComponentType_COMPONENT_TYPE_TIKV {
			clusterConfig.Components.Tikv = &domain.TiKVComponent{
				Compute: domain.Compute{
					CpuNum:    component.CpuNum,
					MemoryGib: component.MemoryGi,
				},
				NodeSize: component.NodeProfileName,
				// StorageType is not required in ListBackups
				StorageType: dataflow.SimpleClusterComponent_STORAGE_TYPE_UNSPECIFIED,
				// RaftIops is not required in ListBackups
				RaftIops:       0,
				StorageSizeGib: component.StorageSizeGi,
				NodeQuantity:   component.Replicas,
			}
		}
		if component.ComponentType == core.ComponentType_COMPONENT_TYPE_TIFLASH {
			clusterConfig.Components.Tiflash = &domain.TiFlashComponent{
				Compute: domain.Compute{
					CpuNum:    component.CpuNum,
					MemoryGib: component.MemoryGi,
				},
				NodeSize: component.NodeProfileName,
				// StorageType is not required in ListBackups
				StorageType:    dataflow.SimpleClusterComponent_STORAGE_TYPE_UNSPECIFIED,
				StorageSizeGib: component.StorageSizeGi,
				NodeQuantity:   component.Replicas,
			}
		}
	})
	return clusterConfig
}

func (b *SnapshotBackupModel) queryClusterSnapshotsForBackups(
	ctx context.Context, backups []entity.Backup,
) (map[uint64]*core.ClusterSnapshot, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	snapshotsMap := make(map[uint64]*core.ClusterSnapshot)
	projectBackups := groupBackupsByProjectID(backups)

	for _, bks := range projectBackups {
		projectSnapshotsMap, err := b.queryProjectClusterSnapshots(
			ctx, bks[0].TenantID.Uint64(), bks[0].ProjectID.Uint64(), bks)
		if err != nil {
			logger.Error("list cluster snapshot failed", zap.Error(err))
			return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
		}
		snapshotsMap = mergeClusterSnapshots(snapshotsMap, projectSnapshotsMap)
	}

	return snapshotsMap, nil
}

func (b *SnapshotBackupModel) queryProjectClusterSnapshots(
	ctx context.Context, tenantID, projectID uint64, backups []entity.Backup,
) (map[uint64]*core.ClusterSnapshot, bizerr.BizError) {
	var snapshotIDs []uint64
	pie.Of(backups).Each(func(backup entity.Backup) {
		if backup.ClusterSnapshotID == 0 {
			return
		}
		snapshotIDs = append(snapshotIDs, uint64(backup.ClusterSnapshotID))
	})
	if len(snapshotIDs) == 0 {
		return nil, nil
	}

	projectSnapshotsMap, err := b.coreCli.ListClusterSnapshot(ctx, tenantID, projectID, snapshotIDs)
	if err != nil {
		return nil, err
	}
	return projectSnapshotsMap, nil
}

func (b *SnapshotBackupModel) preloadBackupScheduleOp() backupRepo.BackupFilterOp {
	return func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.PreloadBackupSchedule()
	}
}

func (b *SnapshotBackupModel) orderByBackupCreateAtDescOp() backupRepo.BackupFilterOp {
	return func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		return qs.OrderDescByCreatedAt()
	}
}

func (b *SnapshotBackupModel) listBackupsFilterOp(filter *domain.ListBackupFilter) backupRepo.BackupFilterOp {
	return func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		if filter == nil {
			return qs
		}
		if filter.TenantID != 0 {
			qs = qs.TenantIDEq(entity.ID(filter.TenantID))
		}
		if filter.ProjectID != 0 {
			qs = qs.ProjectIDEq(entity.ID(filter.ProjectID))
		}
		if len(filter.ClusterIDs) > 0 {
			qs = qs.ClusterIDIn(pie.Map(filter.ClusterIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.BackupIDs) > 0 {
			qs = qs.IDIn(pie.Map(filter.BackupIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.Statuses) > 0 {
			qs = qs.StatusIn(pie.Map(filter.Statuses,
				func(t domain.BackupStatus) entity.BackupStatus { return entity.BackupStatus(t) })...)
		}
		if filter.Region != nil {
			qs = entity.NewBackupQuerySet(qs.GetDB().
				Where("JSON_EXTRACT(bucket_setting_snapshot, '$.buckets[*].regions') like ?",
					fmt.Sprintf("%%%s%%", *filter.Region)))
		}
		qs = qs.DeletedAtIsNull()
		return qs
	}
}

func (b *SnapshotBackupModel) listBackupsPagerOp(pager *domain.Pager) backupRepo.BackupFilterOp {
	if pager == nil {
		pager = domain.NewPager()
	}
	return func(qs entity.BackupQuerySet) entity.BackupQuerySet {
		offset, limit := pager.CalculateOffsetAndLimit()
		qs = qs.Offset(int(offset)).Limit(int(limit))
		return qs
	}
}

func (b *SnapshotBackupModel) preCheckForAsyncDelete(ctx context.Context, backup *entity.Backup) bizerr.BizError {
	if err := b.checkBackupProtected(ctx, backup); err != nil {
		return err
	}
	if err := b.checkFinalBackup(backup); err != nil {
		return err
	}
	if err := b.checkRunningExport(ctx, backup); err != nil {
		return err
	}
	return b.checkRunningRestore(ctx, backup)
}

// checkBackupProtected checks if backup is protected and blocks deletion.
// This is a low-level protection that cannot be bypassed.
func (b *SnapshotBackupModel) checkBackupProtected(ctx context.Context, backup *entity.Backup) bizerr.BizError {
	lg := log.GetLogger(ctx).With(zap.Uint64("backupId", backup.ID.Uint64()))
	if backup.Protected {
		lg.Info("backup is protected and cannot be deleted")
		return bizerr.New(config.ErrCodeBackupProtected, config.ErrMsgBackupProtected, backup.Name())
	}
	return nil
}

// checkRunningExport checks if there are running export tasks for the backup
func (b *SnapshotBackupModel) checkRunningExport(ctx context.Context, backup *entity.Backup) bizerr.BizError {
	total, err := b.backupExportRepo.CountByFilter(ctx, &backupRepo.BackupExportFilter{
		BackupID: pointer.To(backup.ID.Uint64()),
		Statuses: []entity.BackupExportStatus{entity.BackupExportStatusRunning, entity.BackupExportStatusPending},
	})
	if err != nil {
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, backup.ClusterID)
	}
	if total > 0 {
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrMsgAsyncDeleteBackupPreCheckRunningExport, backup.ID)
	}
	return nil
}

func (b *SnapshotBackupModel) checkRunningRestore(ctx context.Context, backup *entity.Backup) bizerr.BizError {
	queryF := func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		return qs.BackupIDEq(backup.ID).
			StatusIn(entity.RestoreStatusPENDING, entity.RestoreStatusRUNNING).
			PreloadBackup().
			OrderAscByCreatedAt()
	}
	total, err := b.restoreRepo.Count(ctx, queryF)
	if err != nil {
		return bizerr_wrapper.NewInternalErr(config.ErrMsgInternalError, backup.ClusterID)
	}
	if total > 0 {
		return bizerr_wrapper.NewInvalidInputError(
			config.ErrMsgAsyncDeleteBackupPreCheckRunningRestore, backup.ID)
	}
	return nil
}

func (*SnapshotBackupModel) checkFinalBackup(backup *entity.Backup) bizerr.BizError {
	if (backup.Status == entity.BackupStatusRunning || backup.Status == entity.BackupStatusPending) &&
		backup.BackupSchedule.Name == internal_config.FinalBackupName {
		return bizerr_wrapper.NewInvalidInputError(config.ErrMsgCanNotDeleteFinalBackup, backup.ClusterID)
	}
	return nil
}

func (b *SnapshotBackupModel) listByScheduleFilter(
	ctx context.Context, req *domain.ListByScheduleFilterReq,
) (*domain.ListBackupResp, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("SnapshotBackupModel.ListByScheduleFilter")
	schedules, err := b.scheduleRepo.List(ctx,
		func(qs entity.BackupScheduleQuerySet) entity.BackupScheduleQuerySet {
			return qs.
				NameEq(req.Filter.Name).
				UserIDEq(entity.ID(req.Filter.UserId)).
				ClusterIDIn(pie.Map(req.Filter.ClusterIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		})
	if err != nil {
		logger.Error("list backup schedule failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}
	if len(schedules) == 0 {
		return &domain.ListBackupResp{Total: 0}, nil
	}

	backups, err := b.backupRepo.List(ctx,
		func(qs entity.BackupQuerySet) entity.BackupQuerySet {
			return qs.PreloadBackupSchedule().
				BackupScheduleIDIn(pie.Map(schedules, func(t entity.BackupSchedule) entity.ID { return t.ID })...)
		})
	if err != nil {
		logger.Error("list backup failed",
			zap.Any("request", req), zap.Any("backupSchedule", schedules), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(config.ErrInternalError)
	}

	if len(backups) == 0 {
		return &domain.ListBackupResp{Total: 0}, nil
	}

	items, err := b.convertListBackupResp(ctx, backups)
	if err != nil {
		return nil, err
	}
	return &domain.ListBackupResp{
		Items: items,
		Total: int64(len(backups)),
	}, nil
}

func (b *SnapshotBackupModel) CreateFinalBackup(
	ctx context.Context, req *domain.CreateFinalBackupReq,
) (backupId uint64, err error) {
	logger := log.GetLogger(ctx).Named("SnapshotBackupModel.CreateFinalBackup")
	cluster, err := b.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, req.ClusterID)
	if err != nil {
		logger.Error("fail to get cluster", zap.Uint64("clusterId", req.ClusterID), zap.Error(err))
		return 0, err
	}
	dc := cluster.GetDedicatedCluster()
	logger.Info("try to create final backup",
		zap.Uint64("clusterId", cluster.GetClusterId()), zap.String("clusterState", dc.GetState().String()))
	request := &account.GetOrCreateSystemUserRequest{
		OrgId: cluster.GetOrganizationId(),
	}
	systemUser, err := b.accountCli.GetOrCreateSystemUser(ctx, request)
	if err != nil {
		logger.Error("fail to GetOrCreateSystemUser", zap.Any("request", req), zap.Error(err))
		return 0, err
	}

	finalBackup, found, err := b.getFinalBackup(ctx, req.ClusterID, systemUser.Id)
	if err != nil {
		logger.Error("fail to getFinalBackup",
			zap.Uint64("clusterId", req.ClusterID), zap.Uint64("systemUserId", systemUser.Id), zap.Error(err))
		return 0, err
	} else if found {
		return finalBackup.ID, nil
	}

	createReq := &domain.AsyncCreateBackupReq{
		Name:       internal_config.FinalBackupName,
		ProjectID:  cluster.GetProjectId(),
		ClusterID:  cluster.GetClusterId(),
		BackupType: dataflow.BackupType_MANUAL,
		UserID:     systemUser.Id,
	}
	createResp, createErr := b.AsyncCreate(ctx, createReq)
	if createErr != nil {
		logger.Error("fail to getFinalBackup", zap.Any("req", createReq), zap.Error(createErr))
		return 0, createErr
	}
	return uint64(createResp.BackupID), nil
}

func (b *SnapshotBackupModel) getFinalBackup(
	ctx context.Context, clusterId uint64, systemUserId uint64,
) (domain.ListBackupRespItem, bool, error) {
	logger := log.GetLogger(ctx).Named("SnapshotBackupModel.getFinalBackup")
	resp, err := b.listByScheduleFilter(ctx,
		&domain.ListByScheduleFilterReq{
			Filter: &domain.ListBackupScheduleFilter{
				ClusterIDs: []uint64{clusterId},
				Name:       internal_config.FinalBackupName,
				UserId:     systemUserId,
			},
		})
	if err != nil {
		logger.Error("fail to listByScheduleFilter",
			zap.Uint64("clusterId", clusterId), zap.Uint64("systemUserId", systemUserId), zap.Error(err))
		return domain.ListBackupRespItem{}, false, err
	}
	if resp.Total == 0 {
		logger.Info("does not get any final backup.",
			zap.Uint64("clusterId", clusterId), zap.Uint64("systemUserId", systemUserId))
		return domain.ListBackupRespItem{}, false, err
	}
	return resp.Items[0], true, nil
}

func (b *SnapshotBackupModel) ListRecycleBinClusters(
	ctx context.Context, req *domain.ListRecycleBinClustersReq,
) (*domain.ListRecycleBinClustersResp, bizerr.BizError) {
	if bizErr := b.checkListRecycleBinClustersReq(ctx, req); bizErr != nil {
		return nil, bizErr
	}
	clusterIDs, bizErr := b.backupRepo.ListClustersWithBackup(ctx, req.ProjectID)
	if bizErr != nil {
		return nil, bizErr.Chain("query backups error")
	}

	clusters, total, bizErr := b.listDeletedClusters(ctx, clusterIDs, req.Pager)
	if bizErr != nil {
		return nil, bizErr.Chain("query clusters error")
	}

	recycleBinClusters := make([]*domain.RecycleBinCluster, 0, len(clusters))
	for _, cluster := range clusters {
		recycleBinClusters = append(recycleBinClusters, domain.Cluster2RecycleBinCluster(cluster))
	}
	return &domain.ListRecycleBinClustersResp{
		Items: recycleBinClusters,
		Total: total,
	}, nil
}

func (b *SnapshotBackupModel) ListRecycleBinClustersV2(
	ctx context.Context, req *domain.ListRecycleBinClustersV2Req,
) (*domain.ListRecycleBinClustersResp, bizerr.BizError) {
	if bizErr := b.checkListRecycleBinClustersV2Req(ctx, req); bizErr != nil {
		return nil, bizErr
	}

	// Get all cluster IDs with backups
	var clusterIDs []uint64
	var bizErr bizerr.BizError
	if len(req.ProjectIDs) > 0 {
		// Query by projects
		clusterIDs, bizErr = b.backupRepo.ListClustersWithBackupByProjects(ctx, req.ProjectIDs)
		if bizErr != nil {
			return nil, bizErr.Chain("query backups by projects error")
		}
	} else {
		// Query by org (all projects under the org)
		clusterIDs, bizErr = b.backupRepo.ListClustersWithBackupByOrg(ctx, req.TenantID)
		if bizErr != nil {
			return nil, bizErr.Chain("query backups by org error")
		}
	}

	clusters, total, bizErr := b.listDeletedClustersV2(ctx, clusterIDs, req)
	if bizErr != nil {
		return nil, bizErr.Chain("query clusters error")
	}

	recycleBinClusters := make([]*domain.RecycleBinCluster, 0, len(clusters))
	for _, cluster := range clusters {
		recycleBinClusters = append(recycleBinClusters, domain.Cluster2RecycleBinCluster(cluster))
	}
	return &domain.ListRecycleBinClustersResp{
		Items: recycleBinClusters,
		Total: total,
	}, nil
}

func (b *SnapshotBackupModel) checkListRecycleBinClustersReq(
	ctx context.Context, req *domain.ListRecycleBinClustersReq,
) bizerr.BizError {
	if req.TenantID == 0 {
		return bizerr_wrapper.NewInvalidInputError("tenant id can't be empty")
	}
	if req.ProjectID == 0 {
		return bizerr_wrapper.NewInvalidInputError("project id can't be empty")
	}

	project, bizErr := b.accountCli.GetProjectByID(ctx, &account.GetProjectByIDRequest{Id: req.ProjectID})
	if bizErr != nil {
		return bizErr.Chain("get project error")
	}
	if project.Project.TenantId != req.TenantID {
		return bizerr_wrapper.NewInvalidInputError("project id and tenant id are not match")
	}
	return nil
}

func (b *SnapshotBackupModel) checkListRecycleBinClustersV2Req(
	ctx context.Context, req *domain.ListRecycleBinClustersV2Req,
) bizerr.BizError {
	if req.TenantID == 0 {
		return bizerr_wrapper.NewInvalidInputError("tenant id can't be empty")
	}

	// ProjectIDs is optional in V2, only check if provided
	if len(req.ProjectIDs) > 0 {
		projects, bizErr := b.accountCli.ListProjects(ctx, &account.ProjectFilter{Ids: req.ProjectIDs})
		if bizErr != nil {
			return bizErr.Chain("get project error")
		}
		for _, project := range projects.Projects {
			if project.TenantId != req.TenantID {
				return bizerr_wrapper.NewInvalidInputError("project id and tenant id are not match")
			}
		}
	}
	return nil
}

func (b *SnapshotBackupModel) listDeletedClusters(
	ctx context.Context, clusterIDs []uint64, pager *domain.Pager,
) ([]*core.Cluster, int64, bizerr.BizError) {
	if len(clusterIDs) == 0 {
		return []*core.Cluster{}, 0, nil
	}

	clustersFilter := &core.ClusterFilter{
		ClusterIds: clusterIDs,
		Deleted:    pointer.ToBool(true),
		IsDevTier:  pointer.ToBool(false),
	}
	countClustersReq := &core.CountClustersReq{
		Filter: clustersFilter,
	}
	total, bizErr := b.coreCli.CountClusters(ctx, countClustersReq)
	if bizErr != nil {
		return nil, 0, bizErr.Chain("count cluster error")
	}

	listClustersReq := &core.ListClustersReq{
		Filter: clustersFilter,
		Sorters: []*base.Sorter{
			{
				Sort:    base.Sort_SORT_DESC,
				OrderBy: "deleted_time",
			},
		},
		Pager: &base.Pager{
			Page:     pager.Page,
			PageSize: pager.PageSize,
		},
	}
	listClustersResp, bizErr := b.coreCli.ListClusters(ctx, listClustersReq)
	if bizErr != nil {
		return nil, 0, bizErr.Chain("list clusters error")
	}

	clusters := pie.Map(listClustersResp.OrderedClusterIds, func(clusterID uint64) *core.Cluster {
		return listClustersResp.Clusters[clusterID]
	})
	return clusters, total, nil
}

func (b *SnapshotBackupModel) listDeletedClustersV2(
	ctx context.Context, clusterIDs []uint64, req *domain.ListRecycleBinClustersV2Req,
) ([]*core.Cluster, int64, bizerr.BizError) {
	// If no clusters have backups, return empty result
	// Recycle bin should only show deleted clusters with backups
	if len(clusterIDs) == 0 {
		return []*core.Cluster{}, 0, nil
	}

	clustersFilter := &core.ClusterFilter{
		ClusterIds: clusterIDs,
		Deleted:    pointer.ToBool(true),
		IsDevTier:  pointer.ToBool(false),
	}

	// Since coreCli.ListClusters pagination doesn't work properly,
	// always fetch all clusters first, then filter and paginate in memory
	allClusters, bizErr := b.fetchAllDeletedClusters(ctx, clustersFilter)
	if bizErr != nil {
		return nil, 0, bizErr
	}

	// Apply filters in memory
	filteredClusters := b.filterClustersV2(allClusters, req)
	total := int64(len(filteredClusters))

	if req.Pager == nil {
		return filteredClusters, total, nil
	}

	// Apply pagination in memory
	paginatedClusters := utils.PaginateSlice(filteredClusters, req.Pager.Page, req.Pager.PageSize)
	return paginatedClusters, total, nil
}

func (b *SnapshotBackupModel) fetchAllDeletedClusters(
	ctx context.Context, filter *core.ClusterFilter,
) ([]*core.Cluster, bizerr.BizError) {
	if len(filter.ClusterIds) == 0 {
		return []*core.Cluster{}, nil
	}

	// Split cluster IDs into smaller groups to avoid timeout
	const chunkSize = 50
	clusterIDGroups := lo.Chunk(filter.ClusterIds, chunkSize)

	// Fetch clusters concurrently with controlled parallelism
	allClusters, err := b.fetchClusterGroupsConcurrently(ctx, filter, clusterIDGroups)
	if err != nil {
		return nil, err
	}

	// Sort by deletion time
	b.sortClustersByDeletionTime(allClusters)
	return allClusters, nil
}

type clusterFetchResult struct {
	clusters []*core.Cluster
	err      bizerr.BizError
	groupIdx int
}

func (b *SnapshotBackupModel) fetchClusterGroupsConcurrently(
	ctx context.Context,
	filter *core.ClusterFilter,
	clusterIDGroups [][]uint64,
) ([]*core.Cluster, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	const maxConcurrency = 10
	sem := semaphore.NewWeighted(maxConcurrency)
	var (
		mu          sync.Mutex
		allClusters []*core.Cluster
		firstError  bizerr.BizError
		wg          sync.WaitGroup
	)
	wg.Add(len(clusterIDGroups))
	for i, groupIDs := range clusterIDGroups {
		go func(groupIdx int, clusterIDs []uint64) {
			defer wg.Done()
			if err := sem.Acquire(ctx, 1); err != nil {
				// Context was canceled before we could acquire the semaphore
				logger.Error("goroutine exiting before acquiring semaphore",
					zap.Int("groupIdx", groupIdx), zap.Error(err))
				return
			}
			defer sem.Release(1)
			select {
			case <-ctx.Done():
				logger.Info("goroutine exiting due to context cancellation",
					zap.Int("groupIdx", groupIdx))
				return
			default:
			}
			groupFilter := &core.ClusterFilter{
				ClusterIds: clusterIDs,
				Deleted:    filter.Deleted,
				IsDevTier:  filter.IsDevTier,
			}
			clusters, bizErr := b.fetchClustersByFilter(ctx, groupFilter, groupIdx)
			mu.Lock()
			defer mu.Unlock()
			if bizErr != nil {
				logger.Error("failed to fetch cluster group",
					zap.Int("groupIdx", groupIdx), zap.Error(bizErr))
				if firstError == nil {
					firstError = bizErr
				}
			} else {
				allClusters = append(allClusters, clusters...)
			}
		}(i, groupIDs)
	}
	wg.Wait()
	if firstError != nil {
		logger.Error("failed to fetch deleted clusters",
			zap.Int("fetchedCount", len(allClusters)), zap.Error(firstError))
		return nil, firstError
	}
	return allClusters, nil
}

func (b *SnapshotBackupModel) sortClustersByDeletionTime(clusters []*core.Cluster) {
	slices.SortFunc(clusters, func(a, b *core.Cluster) int {
		if a.DeletedAt == nil && b.DeletedAt == nil {
			return 0
		}
		if a.DeletedAt == nil {
			return 1
		}
		if b.DeletedAt == nil {
			return -1
		}
		return int(*b.DeletedAt - *a.DeletedAt)
	})
}

// fetchClustersByFilter fetches clusters for a specific filter
func (b *SnapshotBackupModel) fetchClustersByFilter(
	ctx context.Context, filter *core.ClusterFilter, groupIdx int,
) ([]*core.Cluster, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	listClustersReq := &core.ListClustersReq{
		Filter: filter,
		Sorters: []*base.Sorter{
			{
				Sort:    base.Sort_SORT_DESC,
				OrderBy: "deleted_time",
			},
		},
	}
	listClustersResp, bizErr := b.coreCli.ListClusters(ctx, listClustersReq)
	if bizErr != nil {
		logger.Error("failed to list clusters for group",
			zap.Int("groupIdx", groupIdx),
			zap.Int("clusterIDCount", len(filter.ClusterIds)),
			zap.Error(bizErr))
		return nil, bizErr.Chain("list clusters error")
	}
	clusters := pie.Map(listClustersResp.OrderedClusterIds, func(clusterID uint64) *core.Cluster {
		return listClustersResp.Clusters[clusterID]
	})
	return clusters, nil
}

//nolint:gocognit,gocyclo
func (b *SnapshotBackupModel) filterClustersV2(
	clusters []*core.Cluster, req *domain.ListRecycleBinClustersV2Req,
) []*core.Cluster {
	if req.ClusterName == "" && len(req.Regions) == 0 && req.StartTime == nil && req.EndTime == nil {
		return clusters
	}

	clusterNameLower := strings.ToLower(req.ClusterName)
	regionSet := lo.Keyify(req.Regions)

	return lo.Filter(clusters, func(cluster *core.Cluster, _ int) bool {
		if req.ClusterName != "" && !strings.Contains(strings.ToLower(cluster.Name), clusterNameLower) {
			return false
		}
		if len(req.Regions) > 0 {
			if _, ok := regionSet[cluster.RegionName]; !ok {
				return false
			}
		}
		if req.StartTime != nil || req.EndTime != nil {
			if cluster.DeletedAt == nil {
				return false
			}
			deletedTime := time.Unix(*cluster.DeletedAt, 0)
			if req.StartTime != nil && deletedTime.Before(*req.StartTime) {
				return false
			}
			if req.EndTime != nil && deletedTime.After(*req.EndTime) {
				return false
			}
		}
		return true
	})
}

func groupBackupsByProjectID(backups []entity.Backup) map[uint64][]entity.Backup {
	projectBackups := make(map[uint64][]entity.Backup)
	for _, bk := range backups {
		if _, ok := projectBackups[bk.ProjectID.Uint64()]; !ok {
			projectBackups[bk.ProjectID.Uint64()] = make([]entity.Backup, 0, len(backups))
		}
		projectBackups[bk.ProjectID.Uint64()] = append(projectBackups[bk.ProjectID.Uint64()], bk)
	}
	return projectBackups
}

func mergeClusterSnapshots(
	snapshots1, snapshots2 map[uint64]*core.ClusterSnapshot,
) map[uint64]*core.ClusterSnapshot {
	snapshots := make(map[uint64]*core.ClusterSnapshot, len(snapshots1)+len(snapshots2))
	for k, snapshot := range snapshots1 {
		snapshots[k] = snapshot
	}
	for k, snapshot := range snapshots2 {
		snapshots[k] = snapshot
	}
	return snapshots
}