BackupX/server/internal/service/backup_execution_service.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"backupx/server/internal/apperror"
	"backupx/server/internal/backup"
	backupretention "backupx/server/internal/backup/retention"
	"backupx/server/internal/model"
	"backupx/server/internal/repository"
	"backupx/server/internal/storage"
	"backupx/server/internal/storage/codec"
	"backupx/server/pkg/compress"
	backupcrypto "backupx/server/pkg/crypto"
)

type BackupExecutionNotification struct {
	Task   *model.BackupTask
	Record *model.BackupRecord
	Error  error
}

type BackupResultNotifier interface {
	NotifyBackupResult(context.Context, BackupExecutionNotification) error
}

type noopBackupNotifier struct{}

func (noopBackupNotifier) NotifyBackupResult(context.Context, BackupExecutionNotification) error {
	return nil
}

type StorageUploadResultItem struct {
	StorageTargetID   uint   `json:"storageTargetId"`
	StorageTargetName string `json:"storageTargetName"`
	Status            string `json:"status"`
	StoragePath       string `json:"storagePath,omitempty"`
	FileSize          int64  `json:"fileSize,omitempty"`
	Error             string `json:"error,omitempty"`
}

type DownloadedArtifact struct {
	FileName string
	Reader   io.ReadCloser
}

// collectTargetIDs 获取任务关联的所有存储目标 ID
func collectTargetIDs(task *model.BackupTask) []uint {
	if len(task.StorageTargets) > 0 {
		ids := make([]uint, len(task.StorageTargets))
		for i, t := range task.StorageTargets {
			ids[i] = t.ID
		}
		return ids
	}
	if task.StorageTargetID > 0 {
		return []uint{task.StorageTargetID}
	}
	return nil
}

type BackupExecutionService struct {
	tasks           repository.BackupTaskRepository
	records         repository.BackupRecordRepository
	targets         repository.StorageTargetRepository
	storageRegistry *storage.Registry
	runnerRegistry  *backup.Registry
	logHub          *backup.LogHub
	retention       *backupretention.Service
	cipher          *codec.ConfigCipher
	notifier        BackupResultNotifier
	async           func(func())
	now             func() time.Time
	tempDir         string
	semaphore       chan struct{}
}

func NewBackupExecutionService(
	tasks repository.BackupTaskRepository,
	records repository.BackupRecordRepository,
	targets repository.StorageTargetRepository,
	storageRegistry *storage.Registry,
	runnerRegistry *backup.Registry,
	logHub *backup.LogHub,
	retention *backupretention.Service,
	cipher *codec.ConfigCipher,
	notifier BackupResultNotifier,
	tempDir string,
	maxConcurrent int,
) *BackupExecutionService {
	if notifier == nil {
		notifier = noopBackupNotifier{}
	}
	if tempDir == "" {
		tempDir = "/tmp/backupx"
	}
	if maxConcurrent <= 0 {
		maxConcurrent = 2
	}
	return &BackupExecutionService{
		tasks:           tasks,
		records:         records,
		targets:         targets,
		storageRegistry: storageRegistry,
		runnerRegistry:  runnerRegistry,
		logHub:          logHub,
		retention:       retention,
		cipher:          cipher,
		notifier:        notifier,
		async: func(job func()) {
			go job()
		},
		now:       func() time.Time { return time.Now().UTC() },
		tempDir:   tempDir,
		semaphore: make(chan struct{}, maxConcurrent),
	}
}

func (s *BackupExecutionService) RunTaskByID(ctx context.Context, id uint) (*BackupRecordDetail, error) {
	return s.startTask(ctx, id, true)
}

func (s *BackupExecutionService) RunTaskByIDSync(ctx context.Context, id uint) (*BackupRecordDetail, error) {
	return s.startTask(ctx, id, false)
}

func (s *BackupExecutionService) DownloadRecord(ctx context.Context, recordID uint) (*DownloadedArtifact, error) {
	record, provider, err := s.loadRecordProvider(ctx, recordID)
	if err != nil {
		return nil, err
	}
	reader, err := provider.Download(ctx, record.StoragePath)
	if err != nil {
		return nil, apperror.Internal("BACKUP_RECORD_DOWNLOAD_FAILED", "无法下载备份文件", err)
	}
	fileName := record.FileName
	if strings.TrimSpace(fileName) == "" {
		fileName = filepath.Base(record.StoragePath)
	}
	return &DownloadedArtifact{FileName: fileName, Reader: reader}, nil
}

func (s *BackupExecutionService) RestoreRecord(ctx context.Context, recordID uint) error {
	record, provider, err := s.loadRecordProvider(ctx, recordID)
	if err != nil {
		return err
	}
	task, err := s.tasks.FindByID(ctx, record.TaskID)
	if err != nil {
		return apperror.Internal("BACKUP_TASK_GET_FAILED", "无法获取关联备份任务", err)
	}
	if task == nil {
		return apperror.New(404, "BACKUP_TASK_NOT_FOUND", "关联的备份任务不存在，无法执行恢复", fmt.Errorf("backup task %d not found", record.TaskID))
	}
	tempDir, err := os.MkdirTemp("", "backupx-restore-*")
	if err != nil {
		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法创建恢复目录", err)
	}
	defer os.RemoveAll(tempDir)
	artifactPath := filepath.Join(tempDir, filepath.Base(record.FileName))
	if strings.TrimSpace(filepath.Base(record.FileName)) == "" {
		artifactPath = filepath.Join(tempDir, filepath.Base(record.StoragePath))
	}
	reader, err := provider.Download(ctx, record.StoragePath)
	if err != nil {
		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法下载备份文件", err)
	}
	if err := writeReaderToFile(artifactPath, reader); err != nil {
		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法写入恢复文件", err)
	}
	preparedPath, err := s.prepareArtifactForRestore(artifactPath)
	if err != nil {
		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "无法准备恢复文件", err)
	}
	spec, err := s.buildTaskSpec(task, record.StartedAt)
	if err != nil {
		return err
	}
	runner, err := s.runnerRegistry.Runner(spec.Type)
	if err != nil {
		return apperror.BadRequest("BACKUP_TASK_INVALID", "不支持的备份任务类型", err)
	}
	if err := runner.Restore(ctx, spec, preparedPath, backup.NopLogWriter{}); err != nil {
		return apperror.Internal("BACKUP_RECORD_RESTORE_FAILED", "恢复备份失败", err)
	}
	return nil
}

func (s *BackupExecutionService) DeleteRecord(ctx context.Context, recordID uint) error {
	record, provider, err := s.loadRecordProvider(ctx, recordID)
	if err != nil {
		return err
	}
	if strings.TrimSpace(record.StoragePath) != "" {
		if err := provider.Delete(ctx, record.StoragePath); err != nil {
			return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份文件", err)
		}
	}
	if err := s.records.Delete(ctx, recordID); err != nil {
		return apperror.Internal("BACKUP_RECORD_DELETE_FAILED", "无法删除备份记录", err)
	}
	return nil
}

func (s *BackupExecutionService) startTask(ctx context.Context, id uint, async bool) (*BackupRecordDetail, error) {
	task, err := s.tasks.FindByID(ctx, id)
	if err != nil {
		return nil, apperror.Internal("BACKUP_TASK_GET_FAILED", "无法获取备份任务详情", err)
	}
	if task == nil {
		return nil, apperror.New(404, "BACKUP_TASK_NOT_FOUND", "备份任务不存在", fmt.Errorf("backup task %d not found", id))
	}
	startedAt := s.now()
	// 取第一个存储目标 ID 做兼容
	primaryTargetID := task.StorageTargetID
	if tids := collectTargetIDs(task); len(tids) > 0 {
		primaryTargetID = tids[0]
	}
	record := &model.BackupRecord{TaskID: task.ID, StorageTargetID: primaryTargetID, Status: "running", StartedAt: startedAt}
	if err := s.records.Create(ctx, record); err != nil {
		return nil, apperror.Internal("BACKUP_RECORD_CREATE_FAILED", "无法创建备份记录", err)
	}
	task.LastRunAt = &startedAt
	task.LastStatus = "running"
	if err := s.tasks.Update(ctx, task); err != nil {
		return nil, apperror.Internal("BACKUP_TASK_UPDATE_FAILED", "无法更新任务状态", err)
	}
	run := func() {
		s.executeTask(context.Background(), task, record.ID, startedAt)
	}
	if async {
		s.async(run)
	} else {
		run()
	}
	return s.getRecordDetail(ctx, record.ID)
}

func (s *BackupExecutionService) executeTask(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time) {
	s.semaphore <- struct{}{}
	defer func() { <-s.semaphore }()

	logger := backup.NewExecutionLogger(recordID, s.logHub)
	status := "failed"
	errMessage := ""
	var fileName string
	var fileSize int64
	var storagePath string
	var uploadResults []StorageUploadResultItem
	completeRecord := func() {
		if finalizeErr := s.finalizeRecord(ctx, task, recordID, startedAt, status, errMessage, logger.String(), fileName, fileSize, storagePath); finalizeErr != nil {
			logger.Errorf("写回备份记录失败：%v", finalizeErr)
		}
		// 写入多目标上传结果
		if len(uploadResults) > 0 {
			if resultsJSON, marshalErr := json.Marshal(uploadResults); marshalErr == nil {
				if record, findErr := s.records.FindByID(ctx, recordID); findErr == nil && record != nil {
					record.StorageUploadResults = string(resultsJSON)
					_ = s.records.Update(ctx, record)
				}
			}
		}
		if err := s.notifier.NotifyBackupResult(ctx, BackupExecutionNotification{Task: task, Record: &model.BackupRecord{ID: recordID, TaskID: task.ID, Status: status, FileName: fileName, FileSize: fileSize, StoragePath: storagePath, ErrorMessage: errMessage, StartedAt: startedAt}, Error: buildOptionalError(errMessage)}); err != nil {
			logger.Warnf("发送备份通知失败：%v", err)
		}
		s.logHub.Complete(recordID, status)
	}
	defer completeRecord()

	spec, err := s.buildTaskSpec(task, startedAt)
	if err != nil {
		errMessage = err.Error()
		logger.Errorf("构建任务运行时配置失败：%v", err)
		return
	}
	runner, err := s.runnerRegistry.Runner(spec.Type)
	if err != nil {
		errMessage = err.Error()
		logger.Errorf("获取备份执行器失败：%v", err)
		return
	}
	result, err := runner.Run(ctx, spec, logger)
	if err != nil {
		errMessage = err.Error()
		logger.Errorf("执行备份失败：%v", err)
		return
	}
	defer os.RemoveAll(result.TempDir)
	finalPath := result.ArtifactPath
	if strings.EqualFold(task.Compression, "gzip") && !strings.HasSuffix(strings.ToLower(finalPath), ".gz") {
		logger.Infof("开始压缩备份文件")
		compressedPath, compressErr := compress.GzipFile(finalPath)
		if compressErr != nil {
			errMessage = compressErr.Error()
			logger.Errorf("压缩备份文件失败：%v", compressErr)
			return
		}
		finalPath = compressedPath
	}
	if task.Encrypt {
		logger.Infof("开始加密备份文件")
		encryptedPath, encryptErr := backupcrypto.EncryptFile(s.cipher.Key(), finalPath)
		if encryptErr != nil {
			errMessage = encryptErr.Error()
			logger.Errorf("加密备份文件失败：%v", encryptErr)
			return
		}
		finalPath = encryptedPath
	}
	info, err := os.Stat(finalPath)
	if err != nil {
		errMessage = err.Error()
		logger.Errorf("获取备份文件信息失败：%v", err)
		return
	}
	fileSize = info.Size()
	fileName = filepath.Base(finalPath)
	storagePath = backup.BuildStorageKey(task.Type, startedAt, fileName)

	// 收集所有存储目标
	targetIDs := collectTargetIDs(task)
	if len(targetIDs) == 0 {
		errMessage = "没有关联的存储目标"
		logger.Errorf("没有关联的存储目标")
		return
	}

	// 并行上传到所有目标
	uploadResults = make([]StorageUploadResultItem, len(targetIDs))
	var wg sync.WaitGroup
	for i, tid := range targetIDs {
		wg.Add(1)
		go func(index int, targetID uint) {
			defer wg.Done()
			target, findErr := s.targets.FindByID(ctx, targetID)
			targetName := fmt.Sprintf("target-%d", targetID)
			if findErr == nil && target != nil {
				targetName = target.Name
			}
			provider, resolveErr := s.resolveProvider(ctx, targetID)
			if resolveErr != nil {
				uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: resolveErr.Error()}
				logger.Warnf("存储目标 %s 创建客户端失败：%v", targetName, resolveErr)
				return
			}
			artifact, openErr := os.Open(finalPath)
			if openErr != nil {
				uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: openErr.Error()}
				logger.Warnf("存储目标 %s 打开备份文件失败：%v", targetName, openErr)
				return
			}
			defer artifact.Close()
			logger.Infof("开始上传备份到存储目标：%s", targetName)
			if uploadErr := provider.Upload(ctx, storagePath, artifact, fileSize, map[string]string{"taskId": fmt.Sprintf("%d", task.ID), "recordId": fmt.Sprintf("%d", recordID)}); uploadErr != nil {
				uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "failed", Error: uploadErr.Error()}
				logger.Warnf("存储目标 %s 上传失败：%v", targetName, uploadErr)
				return
			}
			uploadResults[index] = StorageUploadResultItem{StorageTargetID: targetID, StorageTargetName: targetName, Status: "success", StoragePath: storagePath, FileSize: fileSize}
			logger.Infof("存储目标 %s 上传成功", targetName)
			// 每个成功目标独立执行保留策略
			if s.retention != nil {
				cleanupResult, cleanupErr := s.retention.Cleanup(ctx, task, provider)
				if cleanupErr != nil {
					logger.Warnf("存储目标 %s 执行保留策略失败：%v", targetName, cleanupErr)
				} else {
					for _, warning := range cleanupResult.Warnings {
						logger.Warnf("存储目标 %s 保留策略警告：%s", targetName, warning)
					}
				}
			}
		}(i, tid)
	}
	wg.Wait()

	// 汇总结果：任意一个 success → 整体 success
	anySuccess := false
	var failedMessages []string
	for _, r := range uploadResults {
		if r.Status == "success" {
			anySuccess = true
		} else if r.Error != "" {
			failedMessages = append(failedMessages, fmt.Sprintf("%s: %s", r.StorageTargetName, r.Error))
		}
	}
	if anySuccess {
		status = "success"
		if len(failedMessages) > 0 {
			logger.Warnf("部分存储目标上传失败：%s", strings.Join(failedMessages, "; "))
		}
		logger.Infof("备份执行完成")
	} else {
		errMessage = strings.Join(failedMessages, "; ")
		logger.Errorf("所有存储目标上传均失败")
	}
}

func (s *BackupExecutionService) finalizeRecord(ctx context.Context, task *model.BackupTask, recordID uint, startedAt time.Time, status string, errorMessage string, logContent string, fileName string, fileSize int64, storagePath string) error {
	record, err := s.records.FindByID(ctx, recordID)
	if err != nil {
		return err
	}
	if record == nil {
		return fmt.Errorf("backup record %d not found", recordID)
	}
	completedAt := s.now()
	record.Status = status
	record.FileName = fileName
	record.FileSize = fileSize
	record.StoragePath = storagePath
	record.DurationSeconds = int(completedAt.Sub(startedAt).Seconds())
	record.ErrorMessage = strings.TrimSpace(errorMessage)
	record.LogContent = strings.TrimSpace(logContent)
	record.CompletedAt = &completedAt
	if err := s.records.Update(ctx, record); err != nil {
		return err
	}
	task.LastRunAt = &startedAt
	task.LastStatus = status
	return s.tasks.Update(ctx, task)
}

func (s *BackupExecutionService) resolveProvider(ctx context.Context, targetID uint) (storage.StorageProvider, error) {
	target, err := s.targets.FindByID(ctx, targetID)
	if err != nil {
		return nil, apperror.Internal("BACKUP_STORAGE_TARGET_GET_FAILED", "无法获取存储目标详情", err)
	}
	if target == nil {
		return nil, apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "关联的存储目标不存在", nil)
	}
	configMap := map[string]any{}
	if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
		return nil, apperror.Internal("BACKUP_STORAGE_TARGET_DECRYPT_FAILED", "无法解密存储目标配置", err)
	}
	provider, err := s.storageRegistry.Create(ctx, target.Type, configMap)
	if err != nil {
		return nil, err
	}
	return provider, nil
}

func (s *BackupExecutionService) buildTaskSpec(task *model.BackupTask, startedAt time.Time) (backup.TaskSpec, error) {
	excludePatterns := []string{}
	if strings.TrimSpace(task.ExcludePatterns) != "" {
		if err := json.Unmarshal([]byte(task.ExcludePatterns), &excludePatterns); err != nil {
			return backup.TaskSpec{}, apperror.Internal("BACKUP_TASK_DECODE_FAILED", "无法解析排除规则", err)
		}
	}
	password := ""
	if strings.TrimSpace(task.DBPasswordCiphertext) != "" {
		plain, err := s.cipher.Decrypt(task.DBPasswordCiphertext)
		if err != nil {
			return backup.TaskSpec{}, apperror.Internal("BACKUP_TASK_DECRYPT_FAILED", "无法解密数据库密码", err)
		}
		password = string(plain)
	}
	sourcePaths := []string{}
	if strings.TrimSpace(task.SourcePaths) != "" {
		if err := json.Unmarshal([]byte(task.SourcePaths), &sourcePaths); err != nil {
			return backup.TaskSpec{}, apperror.Internal("BACKUP_TASK_DECODE_FAILED", "无法解析源路径配置", err)
		}
	}
	return backup.TaskSpec{
		ID:                task.ID,
		Name:              task.Name,
		Type:              task.Type,
		SourcePath:        task.SourcePath,
		SourcePaths:       sourcePaths,
		ExcludePatterns:   excludePatterns,
		StorageTargetID:   task.StorageTargetID,
		StorageTargetType: "",
		Compression:       task.Compression,
		Encrypt:           task.Encrypt,
		RetentionDays:     task.RetentionDays,
		MaxBackups:        task.MaxBackups,
		StartedAt:         startedAt,
		TempDir:           s.tempDir,
		Database: backup.DatabaseSpec{
			Host:     task.DBHost,
			Port:     task.DBPort,
			User:     task.DBUser,
			Password: password,
			Names:    []string{task.DBName},
			Path:     task.DBPath,
		},
	}, nil
}

func (s *BackupExecutionService) loadRecordProvider(ctx context.Context, recordID uint) (*model.BackupRecord, storage.StorageProvider, error) {
	record, err := s.records.FindByID(ctx, recordID)
	if err != nil {
		return nil, nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
	}
	if record == nil {
		return nil, nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
	}
	provider, err := s.resolveProvider(ctx, record.StorageTargetID)
	if err != nil {
		return nil, nil, err
	}
	return record, provider, nil
}

func (s *BackupExecutionService) prepareArtifactForRestore(artifactPath string) (string, error) {
	currentPath := artifactPath
	if strings.HasSuffix(strings.ToLower(currentPath), ".enc") {
		decryptedPath, err := backupcrypto.DecryptFile(s.cipher.Key(), currentPath)
		if err != nil {
			return "", err
		}
		currentPath = decryptedPath
	}
	if strings.HasSuffix(strings.ToLower(currentPath), ".gz") {
		decompressedPath, err := compress.GunzipFile(currentPath)
		if err != nil {
			return "", err
		}
		currentPath = decompressedPath
	}
	return currentPath, nil
}

func (s *BackupExecutionService) getRecordDetail(ctx context.Context, recordID uint) (*BackupRecordDetail, error) {
	record, err := s.records.FindByID(ctx, recordID)
	if err != nil {
		return nil, apperror.Internal("BACKUP_RECORD_GET_FAILED", "无法获取备份记录详情", err)
	}
	if record == nil {
		return nil, apperror.New(404, "BACKUP_RECORD_NOT_FOUND", "备份记录不存在", fmt.Errorf("backup record %d not found", recordID))
	}
	return toBackupRecordDetail(record, s.logHub), nil
}

func writeReaderToFile(targetPath string, reader io.ReadCloser) error {
	defer reader.Close()
	if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil {
		return err
	}
	file, err := os.Create(targetPath)
	if err != nil {
		return err
	}
	defer file.Close()
	_, err = io.Copy(file, reader)
	return err
}

func buildOptionalError(message string) error {
	if strings.TrimSpace(message) == "" {
		return nil
	}
	return fmt.Errorf("%s", message)
}

func buildStorageProviderFromRepos(ctx context.Context, storageTargetID uint, storageTargets repository.StorageTargetRepository, storageRegistry *storage.Registry, cipher *codec.ConfigCipher) (storage.StorageProvider, *model.StorageTarget, error) {
	target, err := storageTargets.FindByID(ctx, storageTargetID)
	if err != nil {
		return nil, nil, apperror.Internal("BACKUP_STORAGE_TARGET_LOOKUP_FAILED", "无法读取存储目标", err)
	}
	if target == nil {
		return nil, nil, apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil)
	}
	var configMap map[string]any
	if err := cipher.DecryptJSON(target.ConfigCiphertext, &configMap); err != nil {
		return nil, nil, apperror.Internal("BACKUP_STORAGE_TARGET_DECRYPT_FAILED", "无法解密存储目标配置", err)
	}
	provider, err := storageRegistry.Create(ctx, storage.ParseProviderType(target.Type), configMap)
	if err != nil {
		return nil, nil, err
	}
	return provider, target, nil
}