package impl

import (
	"context"
	"math"
	"sort"
	"strconv"

	"github.com/AlekSi/pointer"
	"github.com/Masterminds/semver/v3"
	"github.com/elliotchance/pie/v2"
	workflowEntity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	appconfig "github.com/tidbcloud/dataflow-service/app/config"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	"github.com/tidbcloud/dataflow-service/app/internal/status_matrix"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"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"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	globalv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	dedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const hundred = 100

var restoreStatus2DomainMap = map[entity.RestoreStatus]domain.RestoreStatusEnum{
	entity.RestoreStatusPENDING:  domain.RestoreStatusEnumPending,
	entity.RestoreStatusFinished: domain.RestoreStatusEnumSuccess,
	entity.RestoreStatusFAILED:   domain.RestoreStatusEnumFailed,
	entity.RestoreStatusSUCCESS:  domain.RestoreStatusEnumSuccess,
}

var restoreKernelStatus2DomainMap = map[entity.RestoreKernelStatus]domain.RestoreStatusEnum{
	entity.RestoreKernelStatusPending: domain.RestoreStatusEnumPending,
	entity.RestoreKernelStatusRunning: domain.RestoreStatusEnumRunning,
	entity.RestoreKernelStatusFailed:  domain.RestoreStatusEnumFailed,
	entity.RestoreKernelStatusSuccess: domain.RestoreStatusEnumSuccess,
}

type createRestoreParam struct {
	// input
	createRestoreReq *domain.CreateRestoreReqV2

	backup *entity.Backup

	// for 3-layer arch cluster
	createClusterReqV2   *mgmtv1.RestoreDedicatedClusterRequest
	createClusterVersion string
	bucketID             entity.ID

	// just for pitr
	logBackup *entity.LogBackup
}

func (c *createRestoreParam) GetClusterRegion() string {
	return c.createRestoreReq.ClusterConfig.Region
}

type (
	createRestoreParamBuilder func(ctx context.Context, req *createRestoreParam) bizerr.BizError
	clusterSpecValidator      func(ctx context.Context, req *clusterSpecValidatorReq) bizerr.BizError
)

type createRestoreHook interface {
	makeCreateRestoreParam(ctx context.Context, req *domain.CreateRestoreReqV2) (*createRestoreParam, bizerr.BizError)
	makeCreateRestoreEntity(
		ctx context.Context, createParam *createRestoreParam,
	) (*entity.Restore, bizerr.BizError)
	runCreateRestoreDagIns(
		ctx context.Context, createParam *createRestoreParam, restore *entity.Restore,
	) (*workflowEntity.DagInstance, bizerr.BizError)
}

type deleteRestoreHook interface {
	runDeleteRestoreDagIns(
		ctx context.Context, restore *entity.Restore,
	) (*workflowEntity.DagInstance, bizerr.BizError)
}

type clusterSpecValidatorReq struct {
	inputComponents *domain.ClusterComponents
	clusterSnapshot *core.ClusterSnapshot
	backup          *entity.Backup
}

type baseRestoreModel struct {
	createRestoreHook
	deleteRestoreHook
	createRestoreParamBuilders []createRestoreParamBuilder
	clusterSpecValidators      []clusterSpecValidator
	statusMatrix               status_matrix.StatusMatrix

	dbProvider  db.TransProvider
	restoreRepo restore.RestoreRepo
	bucketRepo  repo.BucketRepo

	coreCli               grpcli.CoreGrpcClient
	mgmtSvcCli            grpcli.MgmtSvcGrpcClient
	dedicatedGlobalClient grpcli.DedicatedGlobalGrpcClient
	configCenterCli       grpcli.ConfigCenterGrpcClient

	featureGate feature_gate.FeatureGate
}

func (b *baseRestoreModel) Query(
	ctx context.Context, req *domain.QueryRestoreReqV2,
) (*domain.QueryRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	restore, err := b.restoreRepo.Get(ctx,
		b.preloadBackupOp(),
		b.listRestoresFilterOp(req.Filter),
	)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, bizerr_wrapper.NewNotFoundErr(appconfig.ErrMsgRestoreNotFound)
		}
		logger.Error("query restore failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	status := b.convertRestoreStatusToDomain(restore)
	return &domain.QueryRestoreRespV2{
		Id:               uint64(restore.ID),
		CreateTime:       restore.CreatedAt,
		Status:           status,
		BackupId:         uint64(restore.BackupID),
		BackupClusterId:  uint64(restore.Backup.ClusterID),
		RestoreClusterID: uint64(restore.ClusterID),
		Steps:            b.convertRestoreStepToDomain(restore),
		Progress:         b.convertRestoreProgressToDomain(restore),
	}, nil
}

func (b *baseRestoreModel) convertRestoreStatusToDomain(restore *entity.Restore) domain.RestoreStatusEnum {
	switch restore.Status {
	case entity.RestoreStatusRUNNING:
		if restore.KernelMetrics == nil || restore.KernelMetrics.Status == nil {
			return domain.RestoreStatusEnumRunning
		}
		return restoreKernelStatus2DomainMap[*restore.KernelMetrics.Status]
	case entity.RestoreStatusDeleted, entity.RestoreStatusDeleting:
		if restore.KernelMetrics == nil || restore.KernelMetrics.Status == nil {
			return domain.RestoreStatusEnumTerminated
		}
		status := restoreKernelStatus2DomainMap[*restore.KernelMetrics.Status]
		if status == domain.RestoreStatusEnumSuccess {
			return status
		}
		return domain.RestoreStatusEnumTerminated
	default:
		return restoreStatus2DomainMap[restore.Status]
	}
}

func (b *baseRestoreModel) convertRestoreStepToDomain(restore *entity.Restore) []domain.RestoreStep {
	if restore.KernelMetrics == nil || len(restore.KernelMetrics.Progresses) == 0 {
		return nil
	}
	progresses := restore.KernelMetrics.Progresses
	return pie.Map(progresses, func(progress entity.RestoreKernelProgress) domain.RestoreStep {
		return domain.RestoreStep{
			StepName: progress.Step,
			Progress: float64(progress.Progress),
		}
	})
}

func (b *baseRestoreModel) convertRestoreProgressToDomain(restore *entity.Restore) float64 {
	if restore.KernelMetrics == nil || len(restore.KernelMetrics.Progresses) == 0 {
		return 0
	}
	progresses := restore.KernelMetrics.Progresses
	// the following stepNumber is not a accurate number of steps, the real step number is
	// determined by the restore kernel dynamically.
	stepNumber := 3 // refer to https://docs.pingcap.com/tidb/stable/br-snapshot-guide/#restore-cluster-snapshots
	if restore.Mode == entity.RestoreModePiTR {
		stepNumber = 5 // refer to https://docs.pingcap.com/tidb/stable/br-pitr-guide/#run-pitr
	}
	progress := pie.Sum(pie.Map(progresses, func(progress entity.RestoreKernelProgress) float64 {
		return float64(progress.Progress)
	})) / float64(stepNumber)
	p := math.Round(progress*hundred) / hundred

	// To avoid generate progress above 100 due to the inaccurate step number, we cap it to 100 and lock it to 99 if the restore is not successful.
	if p > hundred {
		p = hundred
	}
	if p > 99.0 && restore.KernelMetrics.Status != nil && *restore.KernelMetrics.Status != entity.RestoreKernelStatusSuccess {
		p = 99.0
	}
	return p
}

func (b *baseRestoreModel) listRestoresFilterOp(filter *domain.ListRestoreFilter) restore.RestoreFilterOp {
	return func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		if len(filter.ProjectIDs) > 0 {
			qs = qs.ProjectIDIn(pie.Map(filter.ProjectIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.TenantIDs) > 0 {
			qs = qs.TenantIDIn(pie.Map(filter.TenantIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.ClusterIDs) > 0 {
			qs = qs.ClusterIDIn(pie.Map(filter.ClusterIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.RestoreIDs) > 0 {
			qs = qs.IDIn(pie.Map(filter.RestoreIDs, func(t uint64) entity.ID { return entity.ID(t) })...)
		}
		if len(filter.BackupClusterIDs) > 0 {
			qs = entity.NewRestoreQuerySet(qs.GetDB().
				Where("backup_id in (select id from backups where cluster_id in (?))", filter.BackupClusterIDs))
		}
		return qs.OrderDescByCreatedAt()
	}
}

func (b *baseRestoreModel) preloadBackupOp() restore.RestoreFilterOp {
	return func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		qs = qs.PreloadBackup()
		return entity.NewRestoreQuerySet(qs.GetDB().Preload("Backup.BackupSchedule"))
	}
}

func (b *baseRestoreModel) List(
	ctx context.Context, req *domain.ListRestoreReqV2,
) (*domain.ListRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	restores, err := b.restoreRepo.List(ctx,
		b.preloadBackupOp(),
		b.listRestoresFilterOp(req.Filter),
		b.listRestoresPagerOp(req.Pager),
	)
	if err != nil {
		logger.Error("list restore failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	count, err := b.restoreRepo.Count(ctx,
		b.listRestoresFilterOp(req.Filter),
	)
	if err != nil {
		return nil, err
	}

	if len(restores) == 0 {
		return &domain.ListRestoreRespV2{Total: count}, nil
	}

	items := pie.Map(restores, func(restore entity.Restore) domain.ListRestoreRespV2Item {
		item := b.convertListRestoreItemToResp(ctx, restore)
		return item
	})
	return &domain.ListRestoreRespV2{
		Items: items,
		Total: count,
	}, nil
}

func (b *baseRestoreModel) convertListRestoreItemToResp(
	ctx context.Context, restore entity.Restore,
) domain.ListRestoreRespV2Item {
	status := b.convertRestoreStatusToDomain(&restore)
	item := domain.ListRestoreRespV2Item{
		Id:                 uint64(restore.ID),
		CreateTime:         restore.CreatedAt.UTC(),
		RestoreClusterName: restore.Name,
		Mode:               domain.RestoreModeEntity2Domain(restore.Mode),
		Status:             status,
		BackupId:           uint64(restore.BackupID),
		Progress:           b.convertRestoreProgressToDomain(&restore),
	}
	if restore.Backup != nil && restore.Backup.ClusterID != 0 {
		item.BackupClusterId = uint64(restore.Backup.ClusterID)
	}
	if restore.Mode == entity.RestoreModeSnapshot && restore.Backup != nil && restore.Backup.BackupSchedule != nil {
		item.BackupName = restore.Backup.BackupSchedule.Name
	}
	if restore.FinishedAt != nil {
		item.FinishedTime = pointer.To(restore.FinishedAt.UTC())
	}
	if restore.ClusterID != 0 {
		item.RestoreClusterID = uint64(restore.ClusterID)
	}
	if restore.Region != nil {
		item.RestoreClusterRegion = *restore.Region
	}
	if restore.Mode == entity.RestoreModeSnapshot {
		item.RestoreTSO = restore.Backup.StartTimeStamp
		item.RestoreTS = b.calculateBackupTs(ctx, *restore.Backup)
	}
	if restore.Mode == entity.RestoreModePiTR {
		item.RestoreTS = pointer.To(restore.RestoredTS)
		item.RestoreTSO = util.TransToTSOStr(restore.RestoredTS)
	}
	return item
}

func (b *baseRestoreModel) calculateBackupTs(ctx context.Context, backup entity.Backup) *int64 {
	logger := log.GetLogger(ctx)
	if backup.StartTimeStamp == "" {
		return nil
	}
	ts, err := util.ParseTSFromTSString(backup.StartTimeStamp)
	if err != nil {
		logger.Warn("parse backup ts failed", zap.String("ts", backup.StartTimeStamp), zap.Error(err))
		return nil
	}
	return pointer.To(ts)
}

func (b *baseRestoreModel) listRestoresPagerOp(pager *domain.Pager) restore.RestoreFilterOp {
	if pager == nil {
		pager = domain.NewPager()
	}
	return func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		offset, limit := pager.CalculateOffsetAndLimit()
		qs = qs.Offset(int(offset)).Limit(int(limit))
		return qs
	}
}

func (b *baseRestoreModel) Create(
	ctx context.Context, req *domain.CreateRestoreReqV2,
) (*domain.CreateRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	if err := req.Validate(); err != nil {
		return nil, err
	}
	createParam, err := b.makeCreateRestoreParam(ctx, req)
	if err != nil {
		return nil, err
	}
	restoreEntity, dagIns, err := b.createRestoreEntity(ctx, createParam)
	if err != nil {
		return nil, err
	}
	// if specify cluster id, means cluster is exist, no need to create
	clusterID := req.ClusterConfig.ClusterID
	if clusterID == 0 {
		newClusterID, bizErr := b.createRestoreCluster(ctx, createParam, restoreEntity)
		if bizErr != nil {
			return nil, bizErr
		}
		clusterID = *newClusterID
	}

	return &domain.CreateRestoreRespV2{
		ID:         uint64(restoreEntity.ID),
		ClusterID:  clusterID,
		WorkflowID: dagIns.ID,
	}, nil
}

func (b *baseRestoreModel) createRestoreCluster(
	ctx context.Context, createParam *createRestoreParam, restoreEntity *entity.Restore,
) (*uint64, bizerr.BizError) {
	logger := log.GetLogger(ctx)

	createClusterReq := createParam.createClusterReqV2
	createClusterReq.RestoreId = uint64(restoreEntity.ID)
	resp, err := b.mgmtSvcCli.RestoreDedicatedCluster(ctx, createClusterReq)
	if err != nil {
		logger.Error("failed to create cluster via mgmt-service",
			zap.Any("request", createClusterReq), zap.Error(err))
		return nil, err
	}
	return pointer.To(resp.ClusterId), nil
}

func (b *baseRestoreModel) createRestoreEntity(
	ctx context.Context, param *createRestoreParam,
) (*entity.Restore, *workflowEntity.DagInstance, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	restoreEntity, err := b.makeCreateRestoreEntity(ctx, param)
	if err != nil {
		return nil, nil, err
	}
	if restoreEntity.ExtraInfo == nil {
		restoreEntity.ExtraInfo = &entity.RestoreExtra{}
	}
	restoreEntity.ExtraInfo.On3LayerArch = true
	var dagIns *workflowEntity.DagInstance
	dbErr := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		err := b.restoreRepo.Save(ctx, restoreEntity)
		if err != nil {
			return nil
		}
		var dagErr error
		dagIns, dagErr = b.runCreateRestoreDagIns(ctx, param, restoreEntity)
		if dagErr != nil {
			return dagErr
		}
		return nil
	})
	if dbErr != nil {
		logger.Error("create restore failed", zap.Error(dbErr))
		return nil, nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return restoreEntity, dagIns, nil
}

func (b *baseRestoreModel) makeCreateRestoreParam(
	ctx context.Context, req *domain.CreateRestoreReqV2,
) (*createRestoreParam, bizerr.BizError) {
	createParam := &createRestoreParam{
		createRestoreReq: req,
	}
	for _, builder := range b.createRestoreParamBuilders {
		if err := builder(ctx, createParam); err != nil {
			return nil, err
		}
	}
	return createParam, nil
}

// bucketParamBuilder resolves and sets the bucket ID on the createRestoreParam.
// If no explicit backup region is provided, it selects the bucket that covers the
// cluster's region, falling back to the backup's default bucket. If a backup region
// is specified, it looks up the corresponding bucket and returns an error if not found.
func (b *baseRestoreModel) bucketParamBuilder(
	ctx context.Context, param *createRestoreParam,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	backupRegion := param.createRestoreReq.BackupRegion

	if backupRegion == "" {
		// No explicit backup region: prefer the bucket covering the cluster's region,
		// falling back to the backup's default bucket.
		clusterRegion := param.GetClusterRegion()
		bucket, found := param.backup.GetBucketByRegion(clusterRegion)
		if found {
			logger.Info("no backup region specified, selected bucket matching cluster region",
				zap.String("clusterRegion", clusterRegion),
				zap.Uint64("bucketID", bucket.BucketId))
			param.bucketID = entity.ID(bucket.BucketId)
		} else {
			logger.Info("no bucket found for cluster region, falling back to backup default bucket",
				zap.String("clusterRegion", clusterRegion),
				zap.Any("defaultBucketID", param.backup.BucketID))
			param.bucketID = param.backup.BucketID
		}
		return nil
	}

	bucket, found := param.backup.GetBucketByRegion(backupRegion)
	if !found {
		logger.Error("no bucket found for the specified backup region",
			zap.Any("bucketSettingSnapshot", param.backup.BucketSettingSnapshot),
			zap.String("backupRegion", backupRegion))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInvalidParam,
			"backupRegion", backupRegion)
	}
	param.bucketID = entity.ID(bucket.BucketId)
	return nil
}

func (b *baseRestoreModel) restoreClusterParamBuilder(
	ctx context.Context, req *createRestoreParam,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if req.createRestoreReq.ClusterConfig.ClusterID != 0 {
		logger.Info("cluster id is specified, skip restoreClusterParamBuilder")
		return nil
	}

	backup := req.backup
	clusterSnapshot, err := b.coreCli.GetClusterSnapshot(
		ctx, backup.ClusterID.Uint64(), uint64(backup.ClusterSnapshotID))
	if err != nil {
		logger.Error("failed to get cluster snapshot",
			zap.Any("cluster snapshot id", backup.ClusterSnapshotID), zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	err = b.executeClusterSpecValidators(ctx, b.clusterSpecValidators, &clusterSpecValidatorReq{
		inputComponents: req.createRestoreReq.ClusterConfig.Components,
		clusterSnapshot: clusterSnapshot,
		backup:          req.backup,
	})
	if err != nil {
		return err
	}

	backupCluster, err := b.mgmtSvcCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(backup.ClusterID))
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	createClusterReq, err := b.buildCreateClusterReqFor3LayerArch(ctx, req, clusterSnapshot, backupCluster)
	if err != nil {
		return err
	}
	req.createClusterReqV2 = createClusterReq

	return nil
}

func (b *baseRestoreModel) executeClusterSpecValidators(
	ctx context.Context, validators []clusterSpecValidator, req *clusterSpecValidatorReq,
) bizerr.BizError {
	for _, validator := range validators {
		if err := validator(ctx, req); err != nil {
			return err
		}
	}
	return nil
}

func (b *baseRestoreModel) buildCreateClusterReqFor3LayerArch(
	ctx context.Context, param *createRestoreParam, snapshot *core.ClusterSnapshot, backupCluster *mgmtv1.Cluster,
) (*mgmtv1.RestoreDedicatedClusterRequest, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	backup := param.backup
	req := param.createRestoreReq
	components := req.ClusterConfig.Components

	createClusterReq := &mgmtv1.RestoreDedicatedClusterRequest{
		CreatedBy:       strconv.FormatUint(req.UserID, 10),
		OrganizationId:  backupCluster.GetOrganizationId(),
		ProjectId:       param.createRestoreReq.ProjectID,
		CloudProvider:   backupCluster.GetCloudProvider(),
		Region:          req.ClusterConfig.Region,
		SourceClusterId: uint64(backup.ClusterID),
		RestoreId:       0, // fill it after the restore entity is created
		ClusterName:     req.ClusterConfig.Name,
		Port:            req.ClusterConfig.Port,
		RootPassword:    req.ClusterConfig.RootPassword,
		TidbComp: &mgmtv1.RestoreDedicatedClusterRequest_ComputeComp{
			SpuName: components.Tidb.NodeSize,
			Count:   int32(components.Tidb.NodeQuantity),
		},
		TikvComp: &mgmtv1.RestoreDedicatedClusterRequest_StorageComp{
			SpuName:       components.Tikv.NodeSize,
			Count:         int32(components.Tikv.NodeQuantity),
			SizeGi:        int32(components.Tikv.StorageSizeGib),
			StorageType:   dedicatedv1beta1.Cluster_StorageNodeSetting_StorageType(components.Tikv.StorageType),
			DataStoreIops: components.Tikv.DataStoreIops,
			RaftIops:      int32(components.Tikv.RaftIops),
		},
		TiflashComp:  nil, // fill it later
		ValidateOnly: false,
		Version:      param.createClusterVersion,
	}
	if tiflash := components.Tiflash; tiflash != nil {
		createClusterReq.TiflashComp = &mgmtv1.RestoreDedicatedClusterRequest_StorageComp{
			SpuName:     tiflash.NodeSize,
			Count:       int32(tiflash.NodeQuantity),
			SizeGi:      int32(tiflash.StorageSizeGib),
			StorageType: dedicatedv1beta1.Cluster_StorageNodeSetting_StorageType(components.Tiflash.StorageType),
		}
	}
	if tiproxy := components.TiProxy; tiproxy != nil {
		createClusterReq.TiproxyComp = &dedicatedv1beta1.TidbNodeGroup_TiProxySetting{
			NodeSpecKey: tiproxy.NodeSize,
			NodeCount:   pointer.To(int32(tiproxy.NodeQuantity)),
		}
	}
	if err := b.overrideByConf(ctx, createClusterReq, param, snapshot, backupCluster); err != nil {
		logger.Error("failed to override cluster config by conf", zap.Error(err))
		return nil, err
	}

	// if req.ClusterConfig.ClusterID is not 0, it means the cluster exists
	// we don't need to check if the cluster can be created
	if req.ClusterConfig.ClusterID == 0 {
		createClusterReq.ValidateOnly = true
		if _, err := b.mgmtSvcCli.RestoreDedicatedCluster(ctx, createClusterReq); err != nil {
			logger.Warn("failed to dryrun create cluster via mgmt-service", zap.Error(err))
			return nil, err
		}
	}
	createClusterReq.ValidateOnly = false
	return createClusterReq, nil
}

func (b *baseRestoreModel) overrideByConf(
	ctx context.Context, createClusterReq *mgmtv1.RestoreDedicatedClusterRequest,
	param *createRestoreParam, snapshot *core.ClusterSnapshot, backupCluster *mgmtv1.Cluster,
) bizerr.BizError {
	backup := param.backup

	override, ok := config.Global().BackupConfig.RestoreOverrides.GetMatched(
		uint64(backup.TenantID), uint64(backup.ProjectID), uint64(backup.ClusterID),
		backupCluster.GetCloudProvider(), backupCluster.GetRegion(),
	)
	if !ok {
		return nil
	}
	if err := b.overrideImages(ctx, override, createClusterReq, backup, backupCluster); err != nil {
		return err
	}
	if err := b.overrideIops(ctx, override, createClusterReq, snapshot, backupCluster); err != nil {
		return err
	}
	return nil

}

func (b *baseRestoreModel) overrideImages(
	ctx context.Context, override config.RestoreOverride, createClusterReq *mgmtv1.RestoreDedicatedClusterRequest,
	backup *entity.Backup, backupCluster *mgmtv1.Cluster,
) bizerr.BizError {
	logger := log.GetLogger(ctx)

	// Apply component image tags
	if len(override.ComponentImageTags) == 0 && len(override.CloneImageTags) == 0 {
		return nil
	}
	canCloneImage, err := b.canCloneSourceImageTag(ctx, backup, backupCluster, override)
	if err != nil {
		return err
	}

	sourceClusterImageTags := make(map[string]string)
	if len(override.CloneImageTags) > 0 && canCloneImage {
		// Get source cluster image tags from config center
		components, err := b.configCenterCli.ListClusterComponents(ctx, uint64(backupCluster.ClusterId))
		if err != nil {
			logger.Error("failed to get source cluster components from config center", zap.Error(err))
			return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
		}

		for _, component := range components {
			if tag := component.GetEffectiveImageTag(); tag != "" {
				sourceClusterImageTags[component.GetComponentName()] = tag
			}
		}
	}
	if clusterConfig := BuildClusterConfigurationWithFallback(
		override.ComponentImageTags,
		override.CloneImageTags,
		sourceClusterImageTags,
	); clusterConfig != nil {
		logger.Info("overriding cluster images",
			zap.Any("componentImageTags", override.ComponentImageTags),
			zap.Strings("cloneImageTags", override.CloneImageTags),
			zap.Any("sourceClusterImageTags", sourceClusterImageTags),
			zap.Any("clusterConfig", clusterConfig.ComponentConfigurations),
		)
		createClusterReq.ClusterConfiguration = clusterConfig
	}
	return nil
}

func (b *baseRestoreModel) overrideIops(
	ctx context.Context, override config.RestoreOverride, createClusterReq *mgmtv1.RestoreDedicatedClusterRequest,
	snapshot *core.ClusterSnapshot, backupCluster *mgmtv1.Cluster,
) bizerr.BizError {
	logger := log.GetLogger(ctx)

	// Get source cluster additional IOPS from cluster components
	dedicatedCluster := backupCluster.GetDedicatedCluster()
	if dedicatedCluster != nil {
		for _, component := range dedicatedCluster.GetClusterComponents() {
			componentType := component.GetComponentType()
			componentName := ""
			switch componentType {
			case globalv1.ComponentType_TIKV:
				componentName = string(entity.TiKV)
			case globalv1.ComponentType_TIFLASH:
				componentName = string(entity.TiFlash)
			default:
				continue
			}

			if pie.Contains(override.CloneAdditionalIops, componentName) {
				// Get additional IOPS from the component's storage
				for name, storage := range component.GetStorages() {
					if name != globalv1.Sku_DATA.String() {
						continue
					}
					additionalIops := storage.GetAdditionalIops()
					if additionalIops <= 0 {
						continue
					}

					logger.Info("overriding additional IOPS for component", zap.String("component", componentName), zap.Uint32("additionalIops", additionalIops))
					if componentName == string(entity.TiKV) {
						createClusterReq.TikvComp.AdditionalIops = int32(additionalIops)
					} else if componentName == string(entity.TiFlash) && createClusterReq.TiflashComp != nil {
						createClusterReq.TiflashComp.AdditionalIops = int32(additionalIops)
					}
				}
			}
		}
	}
	// Apply TiKV additional IOPS
	if override.TiKVDataAdditionalIops > 0 {
		logger.Info("TiKVDataAdditionalIops", zap.Int32("additionalIops", override.TiKVDataAdditionalIops))
		createClusterReq.TikvComp.AdditionalIops = override.TiKVDataAdditionalIops
	}
	return nil
}

func (b *baseRestoreModel) canCloneSourceImageTag(
	ctx context.Context, backup *entity.Backup, backupCluster *mgmtv1.Cluster, override config.RestoreOverride,
) (bool, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	if backup.ClusterVersion == backupCluster.GetDedicatedCluster().Version {
		return true, nil
	}

	if !override.CloneImageTagsIgnorePatchVersionDiff {
		logger.Info("CloneImageTagsIgnorePatchVersionDiff is disabled")
		return false, nil
	}

	backupVersion, err := semver.NewVersion(utils.NormalizeVersion(backup.ClusterVersion))
	if err != nil {
		logger.Error("failed to parse backup cluster version", zap.String("version", backup.ClusterVersion), zap.Error(err))
		return false, bizerr_wrapper.NewInternalErr("invalid backup cluster version: %s", backup.ClusterVersion)
	}
	backupClusterVersion, err := semver.NewVersion(utils.NormalizeVersion(backupCluster.GetDedicatedCluster().Version))
	if err != nil {
		logger.Error("failed to parse backup cluster version", zap.String("version", backupCluster.GetDedicatedCluster().Version), zap.Error(err))
		return false, bizerr_wrapper.NewInternalErr("invalid backup cluster version: %s", backupCluster.GetDedicatedCluster().Version)
	}
	if backupVersion.Major() == backupClusterVersion.Major() && backupVersion.Minor() == backupClusterVersion.Minor() {
		logger.Info("backup cluster version and backup cluster version have the same major and minor version, can clone source image tag")
		return true, nil
	}

	return false, nil
}

//nolint:gomnd
func (b *baseRestoreModel) ComponentTikvChecker(
	ctx context.Context, req *clusterSpecValidatorReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	tikv := req.inputComponents.Tikv
	if err := utils.CheckParamEmptyError("tikv", tikv); err != nil {
		return err
	}
	tikvSnapshot := pie.Of(req.clusterSnapshot.Components).
		Filter(func(c *core.ClusterComponent) bool {
			return c.ComponentType == core.ComponentType_COMPONENT_TYPE_TIKV
		}).
		FirstOr(nil)
	if tikvSnapshot == nil {
		logger.Error("tikv component not found in cluster snapshot")
		return bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}

	backupSizeGi := req.backup.BackupSize / config.GiB
	totalSizeGi := float64(backupSizeGi*3) * config.Global().BackupConfig.RestoreStoragePercentage
	perNodeSizeGi := int64(math.Ceil(totalSizeGi / float64(tikv.NodeQuantity)))
	if tikv.StorageSizeGib < perNodeSizeGi {
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrMsgTikvStorageSizeInvalid, perNodeSizeGi)
	}
	return nil
}

func (b *baseRestoreModel) ComponentTiFlashChecker(
	ctx context.Context, req *clusterSpecValidatorReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	tiflash := req.inputComponents.Tiflash
	if tiflash == nil {
		return nil
	}
	tiflashSnapshot := pie.Of(req.clusterSnapshot.Components).
		Filter(func(c *core.ClusterComponent) bool {
			return c.ComponentType == core.ComponentType_COMPONENT_TYPE_TIFLASH
		}).
		FirstOr(nil)
	if tiflashSnapshot == nil {
		return nil
	}
	if tiflash.StorageSizeGib < tiflashSnapshot.StorageSizeGi {
		logger.Warn("tiflash storage size is less than cluster snapshot")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrMsgComponentStorageSizeInvalid, "tiflash")
	}
	if tiflash.NodeQuantity < tiflashSnapshot.Replicas {
		logger.Warn("tiflash node quantity is less than cluster snapshot")
		return bizerr_wrapper.NewInvalidInputError(appconfig.ErrMsgComponentNodeQuantityInvalid, "tiflash")
	}
	return nil
}

func (b *baseRestoreModel) Cancel(
	ctx context.Context, req *domain.CancelRestoreReqV2,
) (*domain.CancelRestoreRespV2, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx)
	restore, err := b.restoreRepo.Load(ctx, req.RestoreID)
	// Reentrant processing
	if err != nil {
		if bizerr.ErrCodeRecordNotFound == err.ErrCode() {
			return &domain.CancelRestoreRespV2{}, nil
		}
		logger.Error("load restore failed", zap.Any("request", req), zap.Error(err))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	if restore.Status == entity.RestoreStatusDeleting || restore.Status == entity.RestoreStatusDeleted {
		return &domain.CancelRestoreRespV2{}, nil
	}

	isFriendly := b.statusMatrix.IsActionFriendly2Status(
		status_matrix.CurrentStatus{Status: string(restore.Status)},
		string(entity.RestoreActionTypeDelete))
	if !isFriendly {
		logger.Warn("restore status is not friendly",
			zap.Any("request", req), zap.Any("action", "delete"))
		return nil, bizerr_wrapper.NewInvalidInputError(
			appconfig.ErrActionParam, restore.Status, entity.RestoreActionTypeDelete)
	}
	dagIns, err := b.deleteRestoreEntity(ctx, restore)
	if err != nil {
		return nil, err
	}
	return &domain.CancelRestoreRespV2{
		WorkflowID: dagIns.ID,
	}, nil
}

func (b *baseRestoreModel) deleteRestoreEntity(
	ctx context.Context, restore *entity.Restore,
) (*workflowEntity.DagInstance, bizerr.BizError) {
	logger := log.GetLogger(ctx)

	var dagIns *workflowEntity.DagInstance
	dbErr := b.dbProvider.Transaction(ctx, func(ctx context.Context) error {
		// to deleting
		nextAction := b.statusMatrix.CalculateNextStatusForAction(
			status_matrix.CurrentStatus{Status: string(restore.Status)},
			string(entity.RestoreActionTypeDelete))
		restore.Status = entity.RestoreStatus(nextAction.Status)
		err := b.restoreRepo.Save(ctx, restore)
		if err != nil {
			return err
		}

		// start wfs
		var dagErr error
		dagIns, dagErr = b.runDeleteRestoreDagIns(ctx, restore)
		if dagErr != nil {
			return dagErr
		}
		return nil
	})
	if dbErr != nil {
		logger.Error("delete restore failed", zap.Error(dbErr))
		return nil, bizerr_wrapper.NewInternalErr(appconfig.ErrInternalError)
	}
	return dagIns, nil
}

// componentNameToComponentType maps user-friendly component names to proto ComponentType enum.
var componentNameToComponentType = map[string]globalv1.ComponentType{
	"tidb":    globalv1.ComponentType_TIDB,
	"tikv":    globalv1.ComponentType_TIKV,
	"pd":      globalv1.ComponentType_PD,
	"tiflash": globalv1.ComponentType_TIFLASH,
	"tiproxy": globalv1.ComponentType_TIPROXY,
}

// BuildClusterConfiguration converts a map of component name to image tag into a ClusterConfiguration proto.
// Returns nil if the map is empty. Invalid component keys are silently ignored.
func BuildClusterConfiguration(componentImageTags map[string]string) *globalv1.ClusterConfiguration {
	if len(componentImageTags) == 0 {
		return nil
	}

	var configs []*globalv1.ClusterConfiguration_ComponentConfiguration
	for name, tag := range componentImageTags {
		componentType, ok := componentNameToComponentType[name]
		if !ok {
			continue // skip invalid component names
		}
		configs = append(configs, &globalv1.ClusterConfiguration_ComponentConfiguration{
			ComponentType:     componentType,
			OverwriteImageTag: tag,
		})
	}

	if len(configs) == 0 {
		return nil
	}

	// Sort by component type for deterministic output
	sort.Slice(configs, func(i, j int) bool {
		return configs[i].ComponentType < configs[j].ComponentType
	})

	return &globalv1.ClusterConfiguration{
		ComponentConfigurations: configs,
	}
}

// BuildClusterConfigurationWithFallback builds ClusterConfiguration from explicit overrides and source cluster image tags.
// Priority: explicitOverrides > sourceClusterImageTags
func BuildClusterConfigurationWithFallback(
	explicitOverrides map[string]string,
	useSameImageTags []string,
	sourceClusterImageTags map[string]string,
) *globalv1.ClusterConfiguration {
	// Start with source cluster image tags for components in useSameImageTags
	merged := make(map[string]string)
	for _, component := range useSameImageTags {
		if tag, ok := sourceClusterImageTags[component]; ok {
			merged[component] = tag
		}
	}

	// Apply explicit overrides (higher priority)
	for component, tag := range explicitOverrides {
		merged[component] = tag
	}

	return BuildClusterConfiguration(merged)
}