package regionalapi

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	clusterv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/timestamppb"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	sigyaml "sigs.k8s.io/yaml"
)

var (
	unimplementedErr = fmt.Errorf("regional api unimplemented")
)

const (
	createBackupPath      = "/cluster-server/api/v1/clusters/{cluster_id}/snapshotbackups"
	getBackupPath         = "/cluster-server/api/v1/clusters/{cluster_id}/snapshotbackups/{backup_id}"
	deleteBackupPath      = "/cluster-server/api/v1/clusters/{cluster_id}/snapshotbackups/{backup_id}"
	createLogBackupPath   = "/cluster-server/api/v1/clusters/{cluster_id}/logbackups"
	getLogBackupPath      = "/cluster-server/api/v1/clusters/{cluster_id}/logbackups/{backup_id}"
	listLogBackupPath     = "/cluster-server/api/v1/list-all-logbackups"
	stopLogBackupPath     = "/cluster-server/api/v1/clusters/{cluster_id}/logbackups/{backup_id}:stop"
	truncateLogBackupPath = "/cluster-server/api/v1/clusters/{cluster_id}/logbackups/{backup_id}:truncate"
	deleteLogBackupPath   = "/cluster-server/api/v1/clusters/{cluster_id}/logbackups/{backup_id}"
	createRestorePath     = "/cluster-server/api/v1/clusters/{cluster_id}/restores"
	getRestorePath        = "/cluster-server/api/v1/clusters/{cluster_id}/restores/{restore_id}"
	deleteRestorePath     = "/cluster-server/api/v1/clusters/{cluster_id}/restores/{restore_id}"

	rawCreateBackupCRPath = "/cluster-server/api/v1/clusters/{cluster_id}/backup_raw/cr"
	rawGetBackupCRPath    = "/cluster-server/api/v1/clusters/{cluster_id}/backup_raw/cr/{name}"

	clusterIDPathParam    = "cluster_id"
	backupIDPathParam     = "backup_id"
	restoreIDPathParam    = "restore_id"
	backupCRNamePathParam = "name"

	GoMemLimitEnv       = "GOMEMLIMIT"
	BRDefaultGoMemLimit = "1GiB"
)

type brClient struct {
	regionalCli RegionalClient
}

func NewBackupFacade(regionalCli RegionalClient) infraapi.BackupFacade {
	return &brClient{
		regionalCli: regionalCli,
	}
}

func (c *brClient) CreateBackupIgnoreExists(ctx context.Context, cloudProvider, region string, param *infraapi.CreateBackupParam) error {
	logger := log.GetLogger(ctx)
	logger.Info("creating backup with",
		zap.String("instanceType", lo.FromPtr(param.InstanceType)),
		zap.Any("resourceRequirements", param.ResourceRequirements),
		zap.String("backupName", param.BackupName))

	createReq := &clusterv1.CreateSnapshotBackupRequest{
		Backup: &clusterv1.SnapshotBackup{
			Id:                   param.BackupName,
			ClusterId:            fmt.Sprintf("%d", param.ClusterID),
			CommitTs:             param.CommitTS,
			TableFilters:         param.TableFilter,
			InstanceType:         lo.FromPtr(param.InstanceType),
			Configuration:        param.Configuration.Config,
			ImageRepo:            param.Configuration.BaseImage,
			ImageTag:             param.Configuration.Tag,
			Bucket:               param.Storage,
			ResourceRequirements: convertResourceRequirements(param.ResourceRequirements),
			Env:                  setBRDefaultEnv(convertEnvVarsToMap(param.Env)),
		},
	}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(param.ClusterID),
	}
	logicalRegion := &Region{
		TenantID:      param.TenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Post(ctx, logicalRegion, createBackupPath, pathParams, nil, createReq, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusConflict {
			logger.Info("backup already exist", zap.String("backup", param.BackupName))
			return nil
		}
		logger.Error("failed to create backup", zap.String("backup", param.BackupName), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func (c *brClient) GetBackup(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, backupCRName string) (*cloudv1.Backup, error) {
	logger := log.GetLogger(ctx)
	backup := &clusterv1.SnapshotBackup{}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		backupIDPathParam:  backupCRName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Get(ctx, logicalRegion, getBackupPath, pathParams, nil, backup)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			return nil, errors.NotFoundf("object not found")
		}
		logger.Error("failed to get backup", zap.String("backup", backupCRName), zap.Error(bizErr))
		return nil, bizErr
	}
	return &cloudv1.Backup{
		Status: cloudv1.BackupStatus{
			BackupSize: int64(backup.BackupSize),
			BackupTS:   backup.CommitTs,
			Phase:      snapshotBackupStatusToPhase(backup.Status),
		},
	}, nil
}

func snapshotBackupStatusToPhase(status clusterv1.SnapshotBackup_Status) cloudv1.BackupConditionType {
	switch status {
	case clusterv1.SnapshotBackup_COMPLETED:
		return cloudv1.BackupComplete
	case clusterv1.SnapshotBackup_FAILED:
		return cloudv1.BackupFailed
	default:
		return cloudv1.BackupRunning
	}
}

func (c *brClient) ListBackups(ctx context.Context, region string, projectID, clusterID uint64) ([]cloudv1.Backup, error) {
	return nil, unimplementedErr
}

// ListLogBackups returns all log backups from regional 3-layer control plane servers
// and they will be written back to metadb
func (c *brClient) ListLogBackups(ctx context.Context) ([]cloudv1.Backup, error) {
	logger := log.GetLogger(ctx)
	allBackups := []cloudv1.Backup{}

	regions, err := c.regionalCli.ListRegions(ctx)
	if err != nil {
		return nil, err
	}

	for _, region := range regions {
		regionBackups, err := c.listLogBackupsByRegion(ctx, region.Address)
		if err != nil {
			// Skip the logical region if list backup fails
			logger.Warn("failed to list log backups for region", zap.Any("region", region), zap.Error(err))
			continue
		}
		allBackups = append(allBackups, regionBackups...)
	}

	return allBackups, nil
}

func (c *brClient) listLogBackupsByRegion(ctx context.Context, baseURL string) ([]cloudv1.Backup, error) {
	apiURL := baseURL + listLogBackupPath
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil)
	if err != nil {
		return nil, err
	}

	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	data, err := io.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}

	var logBackupResp clusterv1.ListAllLogBackupsResponse
	if err := protojson.Unmarshal(data, &logBackupResp); err != nil {
		return nil, err
	}

	var backups []cloudv1.Backup
	for _, logBackup := range logBackupResp.LogBackups {
		backups = append(backups, cloudv1.Backup{
			ObjectMeta: metav1.ObjectMeta{
				Name: logBackup.Id,
			},
			Status: cloudv1.BackupStatus{
				Phase:                 logBackupStatusToPhase(logBackup.LogStatus),
				BackupPath:            logBackup.Bucket,
				BackupTS:              logBackup.StartTs,
				LogCheckpointTs:       logBackup.CheckpointTs,
				LogTruncateUntil:      logBackup.TruncateUntil,
				LogSubCommandStatuses: pbSubCommandStatusesToCloud(logBackup.LogSubCommandStatuses),
			},
		})
	}

	return backups, nil
}

func logBackupStatusToPhase(status clusterv1.LogBackup_Status) cloudv1.BackupConditionType {
	switch status {
	case clusterv1.LogBackup_FAILED:
		return cloudv1.BackupFailed
	case clusterv1.LogBackup_RUNNING:
		return cloudv1.BackupRunning
	case clusterv1.LogBackup_STOPPED:
		return cloudv1.BackupStopped
	case clusterv1.LogBackup_PREPARE:
		return cloudv1.BackupPrepare
	case clusterv1.LogBackup_SCHEDULED:
		return cloudv1.BackupScheduled
	case clusterv1.LogBackup_TRUNCATING:
		return cloudv1.BackupTruncating
	case clusterv1.LogBackup_TRUNCATED:
		return cloudv1.BackupTruncated
	case clusterv1.LogBackup_STOPPING:
		return cloudv1.BackupStopping
	default:
		return cloudv1.BackupScheduled
	}
}

func pbSubCommandStatusesToCloud(
	statuses map[string]*clusterv1.LogBackup_SubCommandStatus,
) map[cloudv1.LogSubCommandType]cloudv1.LogSubCommandStatus {
	if len(statuses) == 0 {
		return nil
	}

	result := make(map[cloudv1.LogSubCommandType]cloudv1.LogSubCommandStatus, len(statuses))
	for command, status := range statuses {
		if status == nil {
			continue
		}
		result[cloudv1.LogSubCommandType(command)] = cloudv1.LogSubCommandStatus{
			Command:            pbSubCommandTypeToCloud(status.Command),
			TimeStarted:        pbTimestampToMetav1(status.TimeStarted),
			TimeCompleted:      pbTimestampToMetav1(status.TimeCompleted),
			LogTruncatingUntil: status.LogTruncatingUntil,
			Phase:              logBackupStatusToPhase(status.Phase),
			Conditions:         pbSubCommandConditionsToCloud(status.Conditions),
		}
	}
	if len(result) == 0 {
		return nil
	}
	return result
}

func pbSubCommandConditionsToCloud(
	conditions []*clusterv1.LogBackup_SubCommandCondition,
) []cloudv1.BackupCondition {
	if len(conditions) == 0 {
		return nil
	}

	result := make([]cloudv1.BackupCondition, 0, len(conditions))
	for _, condition := range conditions {
		if condition == nil {
			continue
		}
		result = append(result, cloudv1.BackupCondition{
			Command:            pbSubCommandTypeToCloud(condition.Command),
			Type:               logBackupStatusToPhase(condition.Type),
			Status:             pbConditionStatusToCore(condition.Status),
			LastTransitionTime: pbTimestampToMetav1(condition.LastTransitionTime),
			Reason:             condition.Reason,
			Message:            condition.Message,
		})
	}
	if len(result) == 0 {
		return nil
	}
	return result
}

func pbSubCommandTypeToCloud(command clusterv1.LogBackup_SubCommandType) cloudv1.LogSubCommandType {
	switch command {
	case clusterv1.LogBackup_LOG_START:
		return cloudv1.LogStartCommand
	case clusterv1.LogBackup_LOG_TRUNCATE:
		return cloudv1.LogTruncateCommand
	case clusterv1.LogBackup_LOG_STOP:
		return cloudv1.LogStopCommand
	case clusterv1.LogBackup_LOG_PAUSE:
		return cloudv1.LogPauseCommand
	case clusterv1.LogBackup_LOG_RESUME:
		return cloudv1.LogResumeCommand
	default:
		return cloudv1.LogUnknownCommand
	}
}

func pbConditionStatusToCore(status clusterv1.ClusterCondition_ConditionStatus) corev1.ConditionStatus {
	switch status {
	case clusterv1.ClusterCondition_True:
		return corev1.ConditionTrue
	case clusterv1.ClusterCondition_False:
		return corev1.ConditionFalse
	default:
		return corev1.ConditionUnknown
	}
}

func pbTimestampToMetav1(ts *timestamppb.Timestamp) metav1.Time {
	if ts == nil {
		return metav1.Time{}
	}
	return metav1.NewTime(ts.AsTime())
}

func (c *brClient) StartLogBackupIgnoreExists(ctx context.Context, cloudProvider, region string, param *infraapi.BackupParam) error {
	logger := log.GetLogger(ctx)
	logger.Info("starting log backup with",
		zap.String("instanceType", lo.FromPtr(param.InstanceType)),
		zap.Any("resourceRequirements", param.ResourceRequirements),
		zap.String("backupName", param.BackupName))

	logBackup := &clusterv1.CreateLogBackupRequest{
		Backup: &clusterv1.LogBackup{
			Id:                   param.BackupName,
			ClusterId:            fmt.Sprintf("%d", param.ClusterID),
			StartTs:              *param.CommitTS,
			InstanceType:         lo.FromPtr(param.InstanceType),
			Configuration:        param.Configuration.Config,
			ImageRepo:            param.Configuration.BaseImage,
			ImageTag:             param.Configuration.Tag,
			Bucket:               *param.Storage,
			ResourceRequirements: convertResourceRequirements(param.ResourceRequirements),
			Env:                  setBRDefaultEnv(convertEnvVarsToMap(param.Env)),
		},
	}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(param.ClusterID),
	}
	logicalRegion := &Region{
		TenantID:      param.TenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Post(ctx, logicalRegion, createLogBackupPath, pathParams, nil, logBackup, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusConflict {
			logger.Warn("log backup already exist", zap.String("backup", param.BackupName))
			return nil
		}
		logger.Error("failed to start log backup", zap.String("backup", param.BackupName), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func (c *brClient) StopLogBackup(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, logBackupName string) error {
	logger := log.GetLogger(ctx)
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		backupIDPathParam:  logBackupName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Post(ctx, logicalRegion, stopLogBackupPath, pathParams, nil, nil, nil)
	if bizErr != nil {
		logger.Error("failed to stop log backup", zap.String("backup", logBackupName), zap.Error(bizErr))
		return fmt.Errorf("failed to stop log backup: %w", bizErr)
	}
	return nil
}

func (c *brClient) TruncateLogBackup(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, logBackupName, truncateUntil string) error {
	logger := log.GetLogger(ctx)
	truncateReq := &clusterv1.TruncateLogBackupRequest{
		ClusterId: fmt.Sprint(clusterID),
		BackupId:  logBackupName,
		Until:     truncateUntil,
	}
	body, err := protojson.Marshal(truncateReq)
	if err != nil {
		return err
	}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		backupIDPathParam:  logBackupName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Post(ctx, logicalRegion, truncateLogBackupPath, pathParams, nil, body, nil)
	if bizErr != nil {
		logger.Error("failed to truncate log backup", zap.String("backup", logBackupName), zap.Error(bizErr))
		return fmt.Errorf("failed to truncate log backup: %w", bizErr)
	}
	return nil
}

func (c *brClient) DeleteLogBackupIgnoreNotFound(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, logBackupName string) error {
	logger := log.GetLogger(ctx)
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		backupIDPathParam:  logBackupName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Delete(ctx, logicalRegion, deleteLogBackupPath, pathParams, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			return nil
		}
		logger.Error("failed to delete log backup", zap.String("backup", logBackupName), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func (c *brClient) DeleteBackupIgnoreNotFound(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, backupName string) error {
	logger := log.GetLogger(ctx)
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		backupIDPathParam:  backupName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Delete(ctx, logicalRegion, deleteBackupPath, pathParams, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			logger.Info("snapshotbackup not found", zap.String("backup", backupName))
			return nil
		}
		logger.Error("failed to delete backup", zap.String("backup", backupName), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func NewRestoreFacade(regionalCli RegionalClient) infraapi.RestoreFacade {
	return &brClient{
		regionalCli: regionalCli,
	}
}

func (c *brClient) GetRestore(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, restoreName string) (*cloudv1.Restore, error) {
	logger := log.GetLogger(ctx)
	restore := &clusterv1.Restore{}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		restoreIDPathParam: restoreName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Get(ctx, logicalRegion, getRestorePath, pathParams, nil, restore)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			return nil, errors.NotFoundf("object not found")
		}
		logger.Error("failed to get restore", zap.String("restore", restoreName), zap.Error(bizErr))
		return nil, bizErr
	}
	return &cloudv1.Restore{
		Status: cloudv1.RestoreStatus{
			Phase:      restoreStatusToPhase(restore.Status),
			Progresses: convertRestoreProgressesToCloud(restore.Progresses),
		},
	}, nil
}

func restoreStatusToPhase(status clusterv1.Restore_Status) cloudv1.RestoreConditionType {
	switch status {
	case clusterv1.Restore_COMPLETED:
		return cloudv1.RestoreComplete
	case clusterv1.Restore_FAILED:
		return cloudv1.RestoreFailed
	default:
		return cloudv1.RestoreRunning
	}
}

func convertRestoreProgressesToCloud(progresses []*clusterv1.Restore_Progress) []cloudv1.RestoreProgress {
	if len(progresses) == 0 {
		return nil
	}

	result := make([]cloudv1.RestoreProgress, 0, len(progresses))
	for _, progress := range progresses {
		if progress == nil {
			continue
		}
		result = append(result, cloudv1.RestoreProgress{
			Step:               progress.Step,
			Progress:           progress.Progress,
			LastTransitionTime: pbTimestampToMetav1(progress.LastTransitionTime),
		})
	}

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

func (c *brClient) CreateRestoreIgnoreExists(ctx context.Context, cloudProvider, region string, param *infraapi.RestoreParam) error {
	logger := log.GetLogger(ctx)

	logger.Info("creating restore with",
		zap.String("instanceType", lo.FromPtr(param.InstanceType)),
		zap.Any("resourceRequirements", param.ResourceRequirements),
		zap.String("restoreName", param.RestoreName))

	// if pitr restore, set pitrTs
	pitrTs := ptr.Deref(param.RestoredTS, "")

	// If AWS and GCP, we need to construct the URI
	// If Azure, we need to get the SAS token and concatenate it to the URI
	// https://github.com/tidbcloud/dataflow-service/blob/0610606a873d80be9c07b680b403ed00645dc995/app/workflow/action/restore/create/base_running_restore.go#L145-L146
	var snapshotBucket string
	if param.BucketInfo.CloudProvider == "aws" {
		snapshotBucket = fmt.Sprintf("s3://%s/%s", param.BucketInfo.Name, param.BucketInfo.BackupPrefix)
	} else if param.BucketInfo.CloudProvider == "gcp" {
		snapshotBucket = fmt.Sprintf("gs://%s/%s", param.BucketInfo.Name, param.BucketInfo.BackupPrefix)
	} else if param.BucketInfo.CloudProvider == "azure" {
		snapshotBucket = param.BucketInfo.SnapshotBackupStorage
	}

	var logBucket string
	if pitrTs != "" {
		if param.BucketInfo.CloudProvider == "aws" {
			logBucket = fmt.Sprintf("s3://%s/%s", param.BucketInfo.Name, param.BucketInfo.LogBackupPrefix)
		} else if param.BucketInfo.CloudProvider == "gcp" {
			logBucket = fmt.Sprintf("gs://%s/%s", param.BucketInfo.Name, param.BucketInfo.LogBackupPrefix)
		} else if param.BucketInfo.CloudProvider == "azure" {
			logBucket = param.BucketInfo.LogBackupStorage
		}
	}

	var brConfig v1alpha1.BRConfig
	if err := json.Unmarshal([]byte(param.Configuration.Config), &brConfig); err != nil {
		logger.Error("invalid configuration from config-center", zap.String("config", param.Configuration.Config))
		return err
	}

	// param.Options is hardcode in dataflow-service
	// brConfig comes from config-center
	// This breaks single source of truth principle
	// This two parts might have conflict, a better way to handle this is
	// to get all options from config-center
	brConfig.Options = append(brConfig.Options, param.Options...)
	configData, err := json.Marshal(brConfig)
	if err != nil {
		return err
	}

	createReq := &clusterv1.CreateRestoreRequest{
		Restore: &clusterv1.Restore{
			Id:                   param.RestoreName,
			ClusterId:            fmt.Sprintf("%d", param.ClusterID),
			InstanceType:         lo.FromPtr(param.InstanceType),
			Configuration:        string(configData),
			ImageRepo:            param.Configuration.BaseImage,
			ImageTag:             param.Configuration.Tag,
			SnapshotBucket:       snapshotBucket,
			ResourceRequirements: convertResourceRequirements(param.ResourceRequirements),
			Env:                  setBRDefaultEnv(convertEnvVarsToMap(param.Env)),
			// PITR restore
			LogBucket:         logBucket,
			PitrTs:            pitrTs,
			CompactBackupName: param.CompactBackupName,
			// Forward the backup bucket's actual region so BR uses it instead of the cluster region.
			// lo.EmptyableToPtr returns nil for Azure (empty string), preserving the existing fallback.
			BucketRegion: lo.EmptyableToPtr(param.BucketInfo.Region),
		},
	}
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(param.ClusterID),
	}
	logicalRegion := &Region{
		TenantID:      param.TenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	if err := c.regionalCli.Post(ctx, logicalRegion, createRestorePath, pathParams, nil, createReq, nil); err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordExist {
			return nil
		}
		logger.Error("failed to create restore", zap.String("restore", param.RestoreName), zap.Error(err))
		return err
	}
	return nil
}

func (c *brClient) DeleteRestoreIgnoreNotFound(ctx context.Context, cloudProvider, region string, tenantID uint64, projectID uint64, clusterID uint64, restoreName string) error {
	logger := log.GetLogger(ctx)
	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
		restoreIDPathParam: restoreName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Delete(ctx, logicalRegion, deleteRestorePath, pathParams, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			logger.Info("restore not found", zap.String("restore", restoreName))
			return nil
		}
		logger.Error("failed to delete restore", zap.String("restore", restoreName), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func setBRDefaultEnv(envs map[string]string) map[string]string {
	if envs == nil {
		envs = make(map[string]string)
	}
	if _, ok := envs[GoMemLimitEnv]; ok {
		return envs
	}

	envs[GoMemLimitEnv] = BRDefaultGoMemLimit
	return envs
}

func (c *brClient) RawCreateBackupCR(ctx context.Context, cloudProvider, region string, tenantID, projectID, clusterID uint64, backup *cloudv1.Backup) error {
	logger := log.GetLogger(ctx)
	logger.Info("creating raw backup CR on regional",
		zap.String("backupName", backup.Name))

	pathParams := map[string]string{
		clusterIDPathParam: fmt.Sprint(clusterID),
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	req := &clusterv1.RawCreateBackupCRRequest{
		Content: string(lo.Must1(sigyaml.Marshal(backup))),
	}
	bizErr := c.regionalCli.Post(ctx, logicalRegion, rawCreateBackupCRPath, pathParams, nil, req, nil)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusConflict {
			logger.Info("raw backup CR already exists", zap.String("backup", backup.Name))
			return nil
		}
		logger.Error("failed to create raw backup CR", zap.String("backup", backup.Name), zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func (c *brClient) RawGetBackupCR(ctx context.Context, cloudProvider, region string, tenantID, projectID, clusterID uint64, backupName string) (*cloudv1.Backup, error) {
	logger := log.GetLogger(ctx)
	resp := &clusterv1.RawGetBackupCRResponse{}
	backup := &cloudv1.Backup{}
	pathParams := map[string]string{
		clusterIDPathParam:    fmt.Sprint(clusterID),
		backupCRNamePathParam: backupName,
	}
	logicalRegion := &Region{
		TenantID:      tenantID,
		Name:          region,
		CloudProvider: cloudProvider,
	}
	bizErr := c.regionalCli.Get(ctx, logicalRegion, rawGetBackupCRPath, pathParams, nil, resp)
	if bizErr != nil {
		if bizErr.HttpStatus() == http.StatusNotFound {
			return nil, errors.NotFoundf("raw backup CR not found")
		}
		logger.Error("failed to get raw backup CR", zap.String("backup", backupName), zap.Error(bizErr))
		return nil, bizErr
	}
	if err := sigyaml.Unmarshal([]byte(resp.Content), backup); err != nil {
		return nil, fmt.Errorf("unmarshal backup cr yaml: %w", err)
	}
	return backup, nil
}