package compact

import (
	"context"
	"fmt"
	"strconv"

	"github.com/Masterminds/semver/v3"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	backupModel "github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/config_manager"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	apputils "github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/app/workflow"
	"github.com/tidbcloud/dataflow-service/app/workflow/action"
	"github.com/tidbcloud/dataflow-service/internal/config"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"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.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
)

const nilString = "nil"

type CreateCompactBackupTaskAction struct {
	transProvider     db.TransProvider
	compactBackupRepo backup_repo.CompactBackupRepo
	logBackupRepo     backup_repo.LogBackupRepo
	backupSettingRepo backup_repo.BackupSettingRepo
	bucketRepo        repo.BucketRepo
	compactFacade     infraapi.CompactFacade
	mgmtGrpcClient    grpcli.MgmtSvcGrpcClient
	configManager     config_manager.ConfigCenterManager

	compactBackupModel backupModel.CompactBackupModel
}

func NewCreateCompactBackupTaskAction(
	transProvider db.TransProvider,
	compactBackupRepo backup_repo.CompactBackupRepo,
	logBackupRepo backup_repo.LogBackupRepo,
	backupSettingRepo backup_repo.BackupSettingRepo,
	bucketRepo repo.BucketRepo,
	compactFacade infraapi.CompactFacade,
	mgmtGrpcClient grpcli.MgmtSvcGrpcClient,
	configManager config_manager.ConfigCenterManager,
	compactBackupModel backupModel.CompactBackupModel,
) *CreateCompactBackupTaskAction {
	return &CreateCompactBackupTaskAction{
		transProvider:     transProvider,
		compactBackupRepo: compactBackupRepo,
		logBackupRepo:     logBackupRepo,
		backupSettingRepo: backupSettingRepo,
		bucketRepo:        bucketRepo,
		compactFacade:     compactFacade,
		mgmtGrpcClient:    mgmtGrpcClient,
		configManager:     configManager,

		compactBackupModel: compactBackupModel,
	}
}

func (a *CreateCompactBackupTaskAction) Name() string {
	return workflow.ActionCreateCompactBackupTask
}

func (a *CreateCompactBackupTaskAction) Run(ctx run.ExecuteContext, params interface{}) error {
	logger := action.GetLogger(ctx).Named(a.Name())
	stdCtx := log.WithLogger(ctx.Context(), logger)
	logger.Info("[CREATE_ACTION] Starting CreateCompactBackupTaskAction")

	compactBackupID, err := a.getCompactBackupID(ctx, logger)
	if err != nil {
		return err
	}

	// synchronously delete migrations/ before every compact backup task begin to ensure a clean state.
	// NOTE:
	// 1. The compactions/ subdirectory is deleted asynchronously in workflow.ActionDeleteCompactionForPiTR
	// 2. As PiTR is strict to be executed in sequence for the same logbackup (and compact job will only be triggerd in PiTR),
	//  we can safely delete the compactions/ subdirectory without affecting other compact jobs.
	if bErr := a.compactBackupModel.SyncDeleteMigrations(ctx.Context(), compactBackupID); bErr != nil {
		logger.Error("failed to delete migrations/",
			zap.Uint64("compact_backup_id", compactBackupID), zap.Error(bErr))
		return bErr
	}

	return a.transProvider.Transaction(stdCtx, func(transCtx context.Context) error {
		return a.executeCompactBackupWorkflow(transCtx, compactBackupID, logger)
	})
}

func (a *CreateCompactBackupTaskAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
	return nil
}

// getCompactBackupID retrieves and validates the compact backup ID from context
func (a *CreateCompactBackupTaskAction) getCompactBackupID(ctx run.ExecuteContext, logger *zap.Logger) (uint64, error) {
	// Try to get compact backup ID from ShareData first
	compactBackupIDStr, ok := ctx.ShareData().Get(workflow.KeyCompactBackupID)
	if !ok {
		// If not found in ShareData, try to get from DAG variables (set by RunDagWithTags)
		logger.Warn("[CREATE_ACTION] compact backup id not found in share data, trying to get from DAG variables")
		compactBackupIDStr, ok = ctx.GetVar(workflow.KeyCompactBackupID)
		if !ok {
			logger.Error("[CREATE_ACTION] compact backup id not found in DAG variables either")
			return 0, fmt.Errorf("compact backup id not found in share data or DAG variables")
		}
		logger.Info("[CREATE_ACTION] found compact backup id in DAG variables",
			zap.String("compact_backup_id", compactBackupIDStr))

		// Set it in ShareData for subsequent actions to use
		ctx.ShareData().Set(workflow.KeyCompactBackupID, compactBackupIDStr)
		logger.Info("[CREATE_ACTION] set compact backup id in share data for subsequent actions")
	} else {
		logger.Info("[CREATE_ACTION] found compact backup id in share data",
			zap.String("compact_backup_id", compactBackupIDStr))
	}

	compactBackupID, err := strconv.ParseUint(compactBackupIDStr, 10, 64)
	if err != nil {
		logger.Error("invalid compact backup id", zap.String("id", compactBackupIDStr), zap.Error(err))
		return 0, fmt.Errorf("invalid compact backup id: %s", compactBackupIDStr)
	}

	return compactBackupID, nil
}

// executeCompactBackupWorkflow executes the main compact backup workflow within a transaction
func (a *CreateCompactBackupTaskAction) executeCompactBackupWorkflow(
	ctx context.Context,
	compactBackupID uint64,
	logger *zap.Logger,
) error {
	// 1. Load compact backup entity
	compactBackup, err := a.compactBackupRepo.Load(ctx, compactBackupID)
	if err != nil {
		logger.Error("failed to load compact backup", zap.Uint64("id", compactBackupID), zap.Error(err))
		return err
	}

	// 2. Load related log backup
	logBackup, err := a.logBackupRepo.Load(ctx, compactBackup.LogBackupID.Uint64())
	if err != nil {
		logger.Error("failed to load log backup", zap.Uint64("id", compactBackup.LogBackupID.Uint64()), zap.Error(err))
		return err
	}

	// 3. Create compact backup task
	// Get cluster information for region
	cluster, err := a.mgmtGrpcClient.GetAnyDedicatedClusterBasicByID(ctx, compactBackup.ClusterID.Uint64())
	if err != nil {
		logger.Error("failed to get cluster info",
			zap.Uint64("cluster_id", compactBackup.ClusterID.Uint64()), zap.Error(err))
		return err
	}
	region := cluster.GetRegion()
	cloudProvider := cluster.GetCloudProvider()

	// 4. Determine compact backup time range.
	startTSUint, endTSUint, timeRangeErr := a.prepareTimeRange(ctx, compactBackup, logBackup, cluster, logger)
	if timeRangeErr != nil {
		return timeRangeErr
	}

	// Build compact backup parameters
	param, paramErr := a.buildCreateCompactParam(ctx, compactBackup, logBackup, cluster, startTSUint, endTSUint, logger)
	if paramErr != nil {
		return paramErr
	}

	// Execute compact backup creation and update status
	return a.executeCompactBackupCreation(ctx, cloudProvider, region, param, compactBackup, compactBackupID, logger)
}

// prepareTimeRange determines and parses the time range for compact backup.
// If the compactBackup entity already has explicit StartTS/EndTS (e.g. set by PiTR restore),
// those values take precedence over log-backup-derived timestamps.
func (a *CreateCompactBackupTaskAction) prepareTimeRange(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
	cluster *mgmt.Cluster,
	logger *zap.Logger,
) (*uint64, *uint64, error) {
	// Entity-provided TS takes precedence (e.g. set by PiTR restore flow).
	if compactBackup.StartTS != "" || compactBackup.EndTS != "" {
		logger.Info("[PREPARE_TIME_RANGE] using entity-provided time range",
			zap.String("start_ts", compactBackup.StartTS),
			zap.String("end_ts", compactBackup.EndTS))
		var startTSUint, endTSUint *uint64
		if compactBackup.StartTS != "" {
			v, err := strconv.ParseUint(compactBackup.StartTS, 10, 64)
			if err != nil {
				logger.Error("failed to parse entity StartTS", zap.String("start_ts", compactBackup.StartTS), zap.Error(err))
				return nil, nil, err
			}
			startTSUint = &v
		}
		if compactBackup.EndTS != "" {
			v, err := strconv.ParseUint(compactBackup.EndTS, 10, 64)
			if err != nil {
				logger.Error("failed to parse entity EndTS", zap.String("end_ts", compactBackup.EndTS), zap.Error(err))
				return nil, nil, err
			}
			endTSUint = &v
		}
		return startTSUint, endTSUint, nil
	}

	// Fall back to log-backup-derived time range.
	// Use LogBackup's StartTS and CheckpointTS for compact range.
	var startTSUint *uint64
	var endTSUint *uint64

	if logBackup.StartTS != nil && *logBackup.StartTS != "" {
		startTSValue, err := strconv.ParseUint(*logBackup.StartTS, 10, 64)
		if err != nil {
			logger.Error("failed to parse log backup start TS", zap.String("start_ts", *logBackup.StartTS), zap.Error(err))
			return nil, nil, err
		}
		startTSUint = &startTSValue
	}

	// Check if user has configured a minimum start TS
	minStartTS, err := config.Global().CompactConfig.GetCompactMinStartTSWithOverride(
		cluster.GetClusterId(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)
	if err != nil {
		logger.Warn("failed to parse compact min start TS configuration", zap.Error(err))
	} else if minStartTS != nil {
		// Compare with logBackup.StartTS and use the larger value
		if startTSUint == nil || *minStartTS > *startTSUint {
			logger.Info("using configured minimum start TS instead of log backup start TS",
				zap.Uint64("configured_min_start_ts", *minStartTS),
				zap.Uint64p("log_backup_start_ts", startTSUint))
			startTSUint = minStartTS
		} else {
			logger.Info("log backup start TS is already greater than or equal to configured minimum",
				zap.Uint64("configured_min_start_ts", *minStartTS),
				zap.Uint64("log_backup_start_ts", *startTSUint))
		}
	}

	if logBackup.CheckpointTS != "" {
		endTSValue, err := strconv.ParseUint(logBackup.CheckpointTS, 10, 64)
		if err != nil {
			logger.Error("failed to parse log backup checkpoint TS",
				zap.String("checkpoint_ts", logBackup.CheckpointTS), zap.Error(err))
			return nil, nil, err
		}
		endTSUint = &endTSValue
	}

	return startTSUint, endTSUint, nil
}

// buildCreateCompactParam builds the CreateCompactBackupParam with all required configurations
func (a *CreateCompactBackupTaskAction) buildCreateCompactParam(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
	cluster *mgmt.Cluster,
	startTSUint *uint64,
	endTSUint *uint64,
	logger *zap.Logger,
) (*infraapi.CreateCompactBackupParam, error) {
	if cluster == nil {
		return nil, fmt.Errorf("cluster information is required but was nil")
	}

	configData, err := a.prepareCompactBackupConfig(ctx, compactBackup, cluster, logger)
	if err != nil {
		return nil, err
	}
	storageURI, err := a.getStorageURI(ctx, compactBackup, logBackup)
	if err != nil {
		return nil, err
	}

	param := a.constructCompactBackupParam(
		ctx, compactBackup, logBackup, cluster, startTSUint, endTSUint, configData, storageURI)

	// Log the complete parameter object with ResourceRequirements details
	a.logCreateCompactParam(param, logger)

	return param, nil
}

// executeCompactBackupCreation calls the facade API and updates the compact backup status
func (a *CreateCompactBackupTaskAction) executeCompactBackupCreation(
	ctx context.Context,
	cloudProvider string,
	region string,
	param *infraapi.CreateCompactBackupParam,
	compactBackup *entity.CompactBackup,
	compactBackupID uint64,
	logger *zap.Logger,
) error {
	// Call facade to create compact backup using the prepared parameters
	logger.Info("[CREATE_ACTION] Calling regional API to create compact backup",
		zap.Uint64("cluster_id", param.ClusterID),
		zap.String("region", region),
		zap.Uint64("log_backup_id", param.LogBackupID),
		zap.String("storage", param.Storage))

	_, facadeErr := a.compactFacade.CreateCompactBackupIgnoreExists(ctx, cloudProvider, region, param)
	if facadeErr != nil {
		logger.Error("[CREATE_ACTION] failed to create compact backup via regional API",
			zap.Error(facadeErr),
			zap.String("error_type", fmt.Sprintf("%T", facadeErr)))
		compactBackup.Status = entity.CompactBackupStatusFailed
		compactBackup.ErrMsg = facadeErr.Error()
	} else {
		logger.Info("[CREATE_ACTION] compact backup created successfully via regional API",
			zap.String("downstream_compact_backup_id", *compactBackup.DownstreamCompactID))
		compactBackup.Status = entity.CompactBackupStatusRunning
		compactBackup.ErrMsg = ""
	}

	logger.Info("[CREATE_ACTION] Saving compact backup to database",
		zap.Uint64("compact_backup_id", compactBackupID),
		zap.String("status", string(compactBackup.Status)),
		zap.String("downstream_compact_id", a.getDownstreamCompactIDString(compactBackup)))

	if err := a.compactBackupRepo.Save(ctx, compactBackup); err != nil {
		logger.Error("[CREATE_ACTION] failed to update compact backup status",
			zap.Uint64("id", compactBackupID), zap.Error(err))
		return err
	}

	logger.Info("[CREATE_ACTION] compact backup task created successfully",
		zap.Uint64("compact_backup_id", compactBackupID),
		zap.String("final_status", string(compactBackup.Status)),
		zap.String("final_downstream_id", a.getDownstreamCompactIDString(compactBackup)))

	return nil
}

// getDownstreamCompactIDString returns the downstream compact ID as string, or "nil" if not set
func (a *CreateCompactBackupTaskAction) getDownstreamCompactIDString(compactBackup *entity.CompactBackup) string {
	if compactBackup.DownstreamCompactID != nil {
		return *compactBackup.DownstreamCompactID
	}
	return nilString
}

// compactBackupConfigData holds configuration data for compact backup creation
type compactBackupConfigData struct {
	concurrency          uint32
	cleanPolicy          string
	configuration        *infraapi.CompactBackupConfiguration
	resourceRequirements *corev1.ResourceRequirements
}

// prepareCompactBackupConfig prepares configuration data for compact backup creation
func (a *CreateCompactBackupTaskAction) prepareCompactBackupConfig(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	cluster *mgmt.Cluster,
	logger *zap.Logger,
) (*compactBackupConfigData, error) {
	// Prepare create compact backup parameters with reasonable defaults
	concurrency := uint32(config.Global().CompactConfig.CompactBackupConcurrency)
	cleanPolicy := config.Global().CompactConfig.CompactBackupCleanPolicy

	// Get compact backup configuration from config-center
	configuration, errConfig := a.getCompactBackupConfiguration(ctx, compactBackup.ClusterID.Uint64())
	if errConfig != nil {
		logger.Error("failed to get compact backup configuration",
			zap.Uint64("cluster_id", compactBackup.ClusterID.Uint64()), zap.Error(errConfig))
		return nil, errConfig
	}
	logger.Info("using compact backup configuration", zap.Any("configuration", configuration))

	// Apply image tag overrides from CompactOverrides (takes precedence over config-center values).
	brImageTag := config.Global().CompactConfig.GetBRImageTag(
		compactBackup.ClusterID.Uint64(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)
	if brImageTag != "" {
		logger.Info("overriding BR image tag from config", zap.String("br_image_tag", brImageTag))
		configuration.Tag = brImageTag
	}
	tikvImageTag := config.Global().CompactConfig.GetTiKVImageTag(
		compactBackup.ClusterID.Uint64(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)
	if tikvImageTag != "" {
		logger.Info("overriding TiKV image tag from config", zap.String("tikv_image_tag", tikvImageTag))
		configuration.TiKVVersion = tikvImageTag
	}

	// Get resource requirements first and log it
	resourceRequirements := a.getCompactResourceRequirements(cluster)

	return &compactBackupConfigData{
		concurrency:          concurrency,
		cleanPolicy:          cleanPolicy,
		configuration:        configuration,
		resourceRequirements: resourceRequirements,
	}, nil
}

func (a *CreateCompactBackupTaskAction) getStorageURI(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
) (string, error) {
	logger := log.GetLogger(ctx)
	if compactBackup.BucketID == 0 {
		return logBackup.Storage, nil
	}

	// Resolve storage URI: when BucketID is set, load the specified bucket directly.
	bucket, err := a.bucketRepo.Load(ctx, compactBackup.BucketID.Uint64())
	if err != nil {
		return "", fmt.Errorf("failed to load bucket with id %d: %w", compactBackup.BucketID.Uint64(), err)
	}

	uri, uriErr := bucket.URIInBucket(logBackup.Storage)
	if uriErr != nil {
		return "", fmt.Errorf("failed to resolve storage URI in bucket, bucket_id=%d, storage=%s: %w", compactBackup.BucketID.Uint64(), utils.RedactSourceURI(logBackup.Storage), uriErr)
	}
	logger.Info("resolved storage URI for compact backup based on bucket ID",
		zap.Uint64("bucket_id", compactBackup.BucketID.Uint64()),
		zap.String("storage_uri", utils.RedactSourceURI(uri)))
	return uri, nil
}

// constructCompactBackupParam constructs the CreateCompactBackupParam from components
func (a *CreateCompactBackupTaskAction) constructCompactBackupParam(
	ctx context.Context,
	compactBackup *entity.CompactBackup,
	logBackup *entity.LogBackup,
	cluster *mgmt.Cluster,
	startTSUint *uint64,
	endTSUint *uint64,
	configData *compactBackupConfigData,
	storageURI string,
) *infraapi.CreateCompactBackupParam {
	shardCountOverride := config.Global().CompactConfig.GetShardCount(
		cluster.GetClusterId(),
		cluster.GetOrganizationId(),
		cluster.GetProjectId(),
		cluster.GetRegion(),
	)

	return &infraapi.CreateCompactBackupParam{
		// Required parameters
		ClusterID:   compactBackup.ClusterID.Uint64(),
		TenantID:    cluster.GetOrganizationId(),
		ProjectID:   cluster.GetProjectId(),
		LogBackupID: logBackup.ID.Uint64(),
		Storage:     storageURI,
		// Pre-generated deterministic downstream CR name
		CompactBackupName: *compactBackup.DownstreamCompactID,
		// Time range parameters (core business logic)
		StartTS: startTSUint, // Use LogBackup's start timestamp
		EndTS:   endTSUint,   // Use LogBackup's checkpoint timestamp
		// Optional configuration with sensible defaults
		Concurrency: &configData.concurrency,
		InstanceType: config.Global().CompactConfig.GetCompactInstanceType(
			cluster.GetClusterId(),
			cluster.GetOrganizationId(),
			cluster.GetProjectId(),
			cluster.GetCloudProvider(),
			cluster.GetRegion(),
		),
		CleanPolicy:   &configData.cleanPolicy,  // Clean policy from config (default: Retain)
		Configuration: configData.configuration, // Use TiDB/TiKV v8.5.2 configuration
		// Resource configuration (reuse BackupConfig for compact backup)
		ResourceRequirements: configData.resourceRequirements,
		// Monitoring and categorization tags
		TagEnvironmentValue: action.GenInfraCREnvironmentTagValue(),
		TagComponentValue:   "compact-backup", // Component identifier for monitoring
		ShardCount:          computeShardCount(ctx, cluster, shardCountOverride),
		Env:                 []corev1.EnvVar{},
	}
}

// computeShardCount returns the effective shard count for a cluster.
// Returns 0 if the cluster version is older than v8.5.6 (sharded compaction not supported).
// Uses the per-cluster/project/tenant/region override if present;
// otherwise falls back to ceil(tikvCount / 3).
// 0 means no sharding, and any positive integer enables sharded compaction with that shard count.
func computeShardCount(ctx context.Context, cluster *mgmt.Cluster, override int) int {
	logger := log.GetLogger(ctx)
	dc := cluster.GetDedicatedCluster()
	version := dc.GetVersion()
	if !clusterSupportsShardedCompaction(version) {
		logger.Info("cluster version does not support sharded compaction, using shard count 0",
			zap.String("cluster_version", version))
		return 0
	}
	if override >= 0 {
		logger.Info("using shard count override",
			zap.Int("override", override))
		return override
	}
	tikvCount := uint(0)
	for _, comp := range dc.GetClusterComponents() {
		if comp.GetComponentType() == dedicatedv1.ComponentType_TIKV {
			tikvCount = uint(comp.GetCount())
		}
	}
	shardCount := int((tikvCount + 2) / 3)

	logger.Info("computed shard count based on TiKV count",
		zap.Uint("tikv_count", tikvCount),
		zap.Int("shard_count", shardCount))
	return shardCount
}

// clusterSupportsShardedCompaction checks whether the cluster version is >= v8.5.6.
// Pre-release suffixes (e.g. "-beta.1") are stripped before comparison.
func clusterSupportsShardedCompaction(version string) bool {
	v, err := semver.NewVersion(apputils.NormalizeVersion(version))
	if err != nil {
		return false
	}
	return config.Global().BackupConfig.ShardedCompactVersionConstraint.Check(v)
}

// logCreateCompactParam logs detailed information about the CreateCompactBackupParam
func (a *CreateCompactBackupTaskAction) logCreateCompactParam(
	param *infraapi.CreateCompactBackupParam,
	logger *zap.Logger,
) {
	logger.Info("[PARAM_BUILD] Built CreateCompactBackupParam with ResourceRequirements",
		zap.Uint64("cluster_id", param.ClusterID),
		zap.Uint64("tenant_id", param.TenantID),
		zap.Uint64("project_id", param.ProjectID),
		zap.String("instance_type", func() string {
			if param.InstanceType != nil {
				return *param.InstanceType
			}
			return nilString
		}()),
		zap.Any("resource_requirements", param.ResourceRequirements),
		zap.Bool("has_resource_requirements", param.ResourceRequirements != nil),
		zap.Any("resource_requests", func() interface{} {
			if param.ResourceRequirements != nil {
				return param.ResourceRequirements.Requests
			}
			return nil
		}()),
		zap.Any("resource_limits", func() interface{} {
			if param.ResourceRequirements != nil {
				return param.ResourceRequirements.Limits
			}
			return nil
		}()))
}

func (a *CreateCompactBackupTaskAction) ParameterNew() interface{} {
	return nil
}

// getOrInitializeConfig gets or initializes a config component
func (a *CreateCompactBackupTaskAction) getOrInitializeConfig(
	ctx context.Context,
	cluster *mgmt.Cluster,
	component domain.ConfigComponent,
) (*domain.ConfigVersionInfo, bizerr.BizError) {
	logger := log.GetLogger(ctx).Named("getOrInitializeConfig").With(
		zap.Uint64("cluster_id", cluster.GetClusterId()),
		zap.String("component", string(component)))

	// Try to get existing config
	getReq := &domain.GetConfigVersionReq{
		ClusterID: cluster.GetClusterId(),
		Component: component,
	}

	cfgInfo, err := a.configManager.GetConfigVersion(ctx, getReq)
	if err == nil {
		return cfgInfo, nil
	}

	// If not found, initialize it
	if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		logger.Error("get config error", zap.Error(err))
		return nil, err
	}

	logger.Info("config not found, initializing")

	dc := cluster.GetDedicatedCluster()
	// Initialize config
	initReq := &domain.InitConfigVersionReq{
		ClusterID:     cluster.GetClusterId(),
		TenantID:      cluster.GetOrganizationId(),
		Component:     component,
		Version:       dc.GetVersion(),
		CloudProvider: domain.CloudProviderType(cluster.GetCloudProvider()),
	}

	if initErr := a.configManager.InitConfigVersion(ctx, initReq); initErr != nil {
		logger.Warn("init config error", zap.Error(initErr))
		return nil, initErr
	}

	logger.Info("init config successfully")

	// Get the initialized config
	cfgInfo, err = a.configManager.GetConfigVersion(ctx, getReq)
	if err != nil {
		logger.Error("failed to get config after initialization", zap.Error(err))
		return nil, err
	}

	return cfgInfo, nil
}

// getCompactBackupConfiguration retrieves compact backup configuration from config-center
func (a *CreateCompactBackupTaskAction) getCompactBackupConfiguration(
	ctx context.Context, clusterID uint64,
) (*infraapi.CompactBackupConfiguration, error) {
	logger := log.GetLogger(ctx).Named("getCompactBackupConfiguration").With(zap.Uint64("cluster_id", clusterID))

	// Check if config-center is enabled
	if !config.Global().DebugConfig.EnableConfigCenter {
		logger.Info("config center is not enabled, skip")
		// Return empty configuration when config-center is disabled
		return &infraapi.CompactBackupConfiguration{}, nil
	}

	// Get cluster information. All dedicated clusters are 3-layer, so no arch check needed.
	cluster, err := a.mgmtGrpcClient.GetAnyDedicatedClusterBasicByID(ctx, clusterID)
	if err != nil {
		logger.Warn("get cluster error", zap.Error(err))
		return nil, err
	}

	// Get or initialize BR component configuration
	brConfig, brErr := a.getOrInitializeConfig(ctx, cluster, domain.ConfigComponentCompactBackupBR)
	if brErr != nil {
		return nil, brErr
	}

	// Get or initialize TiKV component configuration
	tikvConfig, tikvErr := a.getOrInitializeConfig(ctx, cluster, domain.ConfigComponentCompactBackupTiKV)
	if tikvErr != nil {
		return nil, tikvErr
	}

	return domain.MergeToCompactBackupConfiguration(brConfig, tikvConfig), nil
}

// getCompactResourceRequirements gets resource requirements for compact backup
// Uses dedicated CompactConfig for complete separation from BackupConfig
func (a *CreateCompactBackupTaskAction) getCompactResourceRequirements(
	cluster *mgmt.Cluster,
) *corev1.ResourceRequirements {
	logger := log.GetLogger(context.Background()).Named("getCompactResourceRequirements")

	logger.Info("[RESOURCE_CONFIG] Getting compact resource requirements",
		zap.Uint64("cluster_id", cluster.GetClusterId()),
		zap.Uint64("project_id", cluster.GetProjectId()),
		zap.Uint64("tenant_id", cluster.GetOrganizationId()),
		zap.String("cluster_name", cluster.GetClusterName()),
		zap.String("region", cluster.GetRegion()))

	requirements := config.Global().CompactConfig.GenCompactK8SResourceRequirements(
		cluster.GetClusterId(),
		cluster.GetProjectId(),
		cluster.GetOrganizationId(),
	)

	if requirements != nil {
		logger.Info("[RESOURCE_CONFIG] Found compact resource requirements",
			zap.Any("requests", requirements.Requests),
			zap.Any("limits", requirements.Limits))
	} else {
		logger.Warn("[RESOURCE_CONFIG] No compact resource requirements found, using nil")
	}

	return requirements
}