提供文件操作类utils内容

internal/core/component_initializers.go

@@ -25,6 +25,7 @@ import (
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/lora"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/file"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/token"
 
 	"gorm.io/gorm"
@@ -42,6 +43,8 @@ type Infrastructure struct {
 
 // initInfrastructure 初始化所有基础设施层组件。
 func initInfrastructure(ctx context.Context, cfg *config.Config) (*Infrastructure, error) {
+	file.SetTempRoot(cfg.App.TempPath)
+
 	storage, err := initStorage(ctx, cfg.Database)
 	if err != nil {
 		return nil, err

internal/infra/config/config.go

@@ -4,11 +4,8 @@
 package config
 
 import (
-	"fmt"
-	"os"
-
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
-	"gopkg.in/yaml.v2"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/file"
 )
 
 // Config 代表应用的完整配置结构
@@ -64,6 +61,7 @@ type AppConfig struct {
 	Name      string `yaml:"name"`
 	Version   string `yaml:"version"`
 	JWTSecret string `yaml:"jwt_secret"` // JWT 密钥
+	TempPath  string `yaml:"temp_path"`
 }
 
 // ServerConfig 代表服务器配置
@@ -273,18 +271,7 @@ func NewConfig() *Config {
 
 // Load 从指定路径加载配置文件
 func (c *Config) Load(path string) error {
-	// 读取配置文件
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return fmt.Errorf("配置文件读取失败: %v", err)
-	}
-
-	// 解析YAML配置
-	if err := yaml.Unmarshal(data, c); err != nil {
-		return fmt.Errorf("配置文件解析失败: %v", err)
-	}
-
-	return nil
+	return file.LoadYaml(path, c)
 }
 
 // GenerateAPIKey 用于补齐API Key作为请求头时缺失的部分

internal/infra/repository/ota_repository.go

@@ -12,6 +12,10 @@ import (
 
 // OtaRepository 定义了与 OTA 升级任务相关的数据库操作接口。
 type OtaRepository interface {
+	// Create 创建一个新的 OTA 任务。
+	Create(ctx context.Context, task *models.OTATask) error
+	// FindByID 根据任务 ID 查找任务。
+	FindByID(ctx context.Context, id uint32) (*models.OTATask, error)
 	// FindTasksByStatusesAndCreationTime 根据状态列表和创建时间查找任务。
 	FindTasksByStatusesAndCreationTime(ctx context.Context, statuses []models.OTATaskStatus, createdBefore time.Time) ([]*models.OTATask, error)
 	// Update 更新单个 OTA 任务。
@@ -32,6 +36,20 @@ func NewGormOtaRepository(ctx context.Context, db *gorm.DB) OtaRepository {
 	}
 }
 
+// Create 实现了创建新 OTA 任务的逻辑。
+func (r *gormOtaRepository) Create(ctx context.Context, task *models.OTATask) error {
+	repoCtx := logs.AddFuncName(ctx, r.ctx, "Create")
+	return r.db.WithContext(repoCtx).Create(task).Error
+}
+
+// FindByID 实现了根据 ID 查找任务的逻辑。
+func (r *gormOtaRepository) FindByID(ctx context.Context, id uint32) (*models.OTATask, error) {
+	repoCtx := logs.AddFuncName(ctx, r.ctx, "FindByID")
+	var task models.OTATask
+	err := r.db.WithContext(repoCtx).First(&task, id).Error
+	return &task, err
+}
+
 // FindTasksByStatusesAndCreationTime 实现了根据状态和创建时间查找任务的逻辑。
 func (r *gormOtaRepository) FindTasksByStatusesAndCreationTime(ctx context.Context,
 	statuses []models.OTATaskStatus,

internal/infra/utils/file/decompress.go

@@ -0,0 +1,243 @@
+package file
+
+import (
+	"archive/tar"
+	"archive/zip"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/bodgit/sevenzip"
+	"github.com/dsnet/compress/bzip2"
+)
+
+// CompressionType 定义了支持的压缩文件类型枚举。
+// 每个枚举值都是一个唯一的文件后缀。
+type CompressionType string
+
+const (
+	ZIP     CompressionType = ".zip"
+	TAR     CompressionType = ".tar"
+	TARGZ   CompressionType = ".tar.gz"
+	TGZ     CompressionType = ".tgz"
+	TARBZ2  CompressionType = ".tar.bz2"
+	TBZ2    CompressionType = ".tbz2"
+	SEVEN_Z CompressionType = ".7z"
+)
+
+// Matches 判断文件名是否以当前压缩类型定义的后缀结尾。
+func (ct CompressionType) Matches(filename string) bool {
+	return strings.HasSuffix(strings.ToLower(filename), string(ct))
+}
+
+// DecompressAtomic 以原子方式将指定的压缩包解压到目标目录。
+// 此函数是线程安全的，它会获取全局文件锁。
+//
+// 核心保障：如果解压过程中发生任何错误，它会自动删除整个 destPath 目录以进行回滚。
+//
+// sourcePath: 要解压的压缩包的完整路径。
+// destPath:   目标解压目录。此函数将创建此目录，并将所有内容解压到其中。
+//
+// 重要警告：请勿将一个已存在的、包含重要文件的目录作为 destPath。
+// 此函数假定它拥有对 destPath 的完全控制权，并会在失败时将其彻底删除。
+func DecompressAtomic(sourcePath, destPath string) error {
+	action := func() error {
+		return decompress(sourcePath, destPath)
+	}
+
+	onRollback := func(err error) {
+		_ = os.RemoveAll(destPath)
+	}
+
+	return ExecuteWithLock(action, onRollback)
+}
+
+// decompress 是解压操作的核心实现，它本身不是线程安全的，也不提供回滚。
+// 它应该被 DecompressAtomic 或其他调用方在 ExecuteWithLock 的回调中执行。
+func decompress(sourcePath, destPath string) error {
+	// 确保目标目录存在
+	if err := os.MkdirAll(destPath, 0755); err != nil {
+		return fmt.Errorf("创建目标目录 %s 失败: %w", destPath, err)
+	}
+
+	// 直接在 switch 中使用 Matches 方法
+	switch {
+	case ZIP.Matches(sourcePath):
+		return decompressZip(sourcePath, destPath)
+	case TAR.Matches(sourcePath),
+		TARGZ.Matches(sourcePath),
+		TGZ.Matches(sourcePath),
+		TARBZ2.Matches(sourcePath),
+		TBZ2.Matches(sourcePath):
+		return decompressTar(sourcePath, destPath)
+	case SEVEN_Z.Matches(sourcePath):
+		return decompress7z(sourcePath, destPath)
+	default:
+		return fmt.Errorf("不支持的压缩文件类型: %s", sourcePath)
+	}
+}
+
+// secureJoinPath 安全地将子路径连接到目标目录，并检查路径遍历攻击。
+func secureJoinPath(destDir, subPath string) (string, error) {
+	destFilePath := filepath.Join(destDir, subPath)
+	if !strings.HasPrefix(destFilePath, filepath.Clean(destDir)+string(os.PathSeparator)) {
+		return "", fmt.Errorf("检测到不安全的解压路径: %s", subPath)
+	}
+	return destFilePath, nil
+}
+
+// createFileFromReader 从一个 reader 创建并写入文件内容。
+func createFileFromReader(filePath string, mode os.FileMode, reader io.Reader) error {
+	// 确保父目录存在
+	if err := os.MkdirAll(filepath.Dir(filePath), 0755); err != nil {
+		return fmt.Errorf("为文件 %s 创建父目录失败: %w", filePath, err)
+	}
+
+	// 创建目标文件
+	destFile, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, mode)
+	if err != nil {
+		return fmt.Errorf("创建目标文件 %s 失败: %w", filePath, err)
+	}
+	defer destFile.Close()
+
+	// 复制内容
+	if _, err = io.Copy(destFile, reader); err != nil {
+		return fmt.Errorf("写入文件 %s 内容失败: %w", filePath, err)
+	}
+
+	return nil
+}
+
+// decompressZip 负责解压 ZIP 格式的压缩包。
+func decompressZip(sourcePath, destPath string) error {
+	reader, err := zip.OpenReader(sourcePath)
+	if err != nil {
+		return fmt.Errorf("打开 ZIP 文件 %s 失败: %w", sourcePath, err)
+	}
+	defer reader.Close()
+
+	for _, f := range reader.File {
+		destFilePath, err := secureJoinPath(destPath, f.Name)
+		if err != nil {
+			return err
+		}
+
+		if f.FileInfo().IsDir() {
+			if err := os.MkdirAll(destFilePath, f.Mode()); err != nil {
+				return fmt.Errorf("创建 ZIP 内目录 %s 失败: %w", destFilePath, err)
+			}
+			continue
+		}
+
+		srcFile, err := f.Open()
+		if err != nil {
+			return fmt.Errorf("打开 ZIP 内文件 %s 失败: %w", f.Name, err)
+		}
+		defer srcFile.Close()
+
+		if err := createFileFromReader(destFilePath, f.Mode(), srcFile); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// processTarStream 是处理 tar 流的核心逻辑。
+func processTarStream(tarReader *tar.Reader, destPath string) error {
+	for {
+		header, err := tarReader.Next()
+		if err == io.EOF {
+			break // 归档结束
+		}
+		if err != nil {
+			return fmt.Errorf("读取 tar 头部失败: %w", err)
+		}
+
+		destFilePath, err := secureJoinPath(destPath, header.Name)
+		if err != nil {
+			return err
+		}
+
+		switch header.Typeflag {
+		case tar.TypeDir:
+			if err := os.MkdirAll(destFilePath, os.FileMode(header.Mode)); err != nil {
+				return fmt.Errorf("创建 tar 内目录 %s 失败: %w", destFilePath, err)
+			}
+		case tar.TypeReg:
+			if err := createFileFromReader(destFilePath, os.FileMode(header.Mode), tarReader); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// decompressTar 负责解压 TAR 格式的归档文件。
+func decompressTar(sourcePath, destPath string) error {
+	file, err := os.Open(sourcePath)
+	if err != nil {
+		return fmt.Errorf("打开 tar 文件 %s 失败: %w", sourcePath, err)
+	}
+	defer file.Close()
+
+	var streamReader io.Reader = file
+	lowerPath := strings.ToLower(sourcePath)
+
+	if TARGZ.Matches(lowerPath) || TGZ.Matches(lowerPath) {
+		gzReader, err := gzip.NewReader(file)
+		if err != nil {
+			return fmt.Errorf("创建 gzip 读取器失败: %w", err)
+		}
+		defer gzReader.Close()
+		streamReader = gzReader
+	} else if TARBZ2.Matches(lowerPath) || TBZ2.Matches(lowerPath) {
+		bz2Reader, err := bzip2.NewReader(file, nil)
+		if err != nil {
+			return fmt.Errorf("创建 bzip2 读取器失败: %w", err)
+		}
+		defer bz2Reader.Close()
+		streamReader = bz2Reader
+	}
+
+	tarReader := tar.NewReader(streamReader)
+	return processTarStream(tarReader, destPath)
+}
+
+// decompress7z 负责解压 7z 格式的压缩包。
+func decompress7z(sourcePath, destPath string) error {
+	reader, err := sevenzip.OpenReader(sourcePath)
+	if err != nil {
+		return fmt.Errorf("打开 7z 文件 %s 失败: %w", sourcePath, err)
+	}
+	defer reader.Close()
+
+	for _, f := range reader.File {
+		destFilePath, err := secureJoinPath(destPath, f.Name)
+		if err != nil {
+			return err
+		}
+
+		if f.FileInfo().IsDir() {
+			if err := os.MkdirAll(destFilePath, f.Mode()); err != nil {
+				return fmt.Errorf("创建 7z 内目录 %s 失败: %w", destFilePath, err)
+			}
+			continue
+		}
+
+		srcFile, err := f.Open()
+		if err != nil {
+			return fmt.Errorf("打开 7z 内文件 %s 失败: %w", f.Name, err)
+		}
+		defer srcFile.Close()
+
+		if err := createFileFromReader(destFilePath, f.Mode(), srcFile); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

internal/infra/utils/file/file.go

@@ -0,0 +1,133 @@
+package file
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"gopkg.in/yaml.v2"
+)
+
+// manager 是文件管理器的内部实现结构体，用于管理临时文件根路径和提供并发安全。
+type manager struct {
+	tempRoot string     // 临时文件存储的根路径
+	mu       sync.Mutex // 用于保证文件操作的线程安全
+}
+
+var (
+	instance *manager
+	once     sync.Once
+)
+
+// init 在包被导入时自动执行，用于初始化文件管理器单例。
+func init() {
+	once.Do(func() {
+		instance = &manager{
+			tempRoot: "./tmp", // 默认的临时文件存储根路径
+		}
+	})
+}
+
+// SetTempRoot 设置临时文件存储的根目录。
+// 此函数是线程安全的。
+func SetTempRoot(path string) {
+	instance.mu.Lock()
+	defer instance.mu.Unlock()
+	instance.tempRoot = path
+}
+
+// ExecuteWithLock 提供一个通用的、带回滚能力的事务性执行单元。
+// 它会获取一个全局锁，然后执行 action。如果 action 执行出错，它将调用 onRollback。
+//
+// action: 需要以原子方式执行的操作。
+// onRollback: 当 action 返回错误时执行的回滚操作，原始错误会传入其中。
+func ExecuteWithLock(action func() error, onRollback func(err error)) error {
+	instance.mu.Lock()
+	defer instance.mu.Unlock()
+
+	err := action()
+	if err != nil && onRollback != nil {
+		onRollback(err)
+	}
+	return err
+}
+
+// CreateTempDir 在指定的根目录下创建一个子目录。
+// 注意：此函数本身不是线程安全的，应在 ExecuteWithLock 的回调中使用。
+func CreateTempDir(tempRoot, subDir string) (string, error) {
+	fullDirPath := filepath.Join(tempRoot, subDir)
+	if err := os.MkdirAll(fullDirPath, 0755); err != nil {
+		return "", fmt.Errorf("创建临时目录 %s 失败: %w", fullDirPath, err)
+	}
+	return fullDirPath, nil
+}
+
+// PrepareTempFilePath 获取文件在指定子目录下的完整路径，并确保父目录存在。
+// 注意：此函数本身不是线程安全的，应在 ExecuteWithLock 的回调中使用。
+func PrepareTempFilePath(tempRoot, subDir, fileName string) (string, error) {
+	fullDirPath := filepath.Join(tempRoot, subDir)
+	if err := os.MkdirAll(fullDirPath, 0755); err != nil {
+		return "", fmt.Errorf("创建临时文件父目录 %s 失败: %w", fullDirPath, err)
+	}
+	return filepath.Join(fullDirPath, fileName), nil
+}
+
+// RemoveTempDir 在指定的根目录下清理并删除一个子目录。
+// 注意：此函数本身不是线程安全的，应在 ExecuteWithLock 的回调中使用。
+func RemoveTempDir(tempRoot, subDir string) error {
+	fullDirPath := filepath.Join(tempRoot, subDir)
+	return os.RemoveAll(fullDirPath)
+}
+
+// RemoveDir 清理并删除指定的任意目录。
+// 此函数与临时文件管理器无关，本身是线程安全的（依赖于操作系统的原子性）。
+func RemoveDir(dirPath string) error {
+	return os.RemoveAll(dirPath)
+}
+
+// LoadYaml 从指定路径加载并解析一个 YAML 文件到传入的结构体中。
+// path: YAML 文件的路径。
+// c: 目标结构体的指针，例如 &MyConfig{}。
+func LoadYaml(path string, c interface{}) error {
+	// 读取配置文件
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return fmt.Errorf("配置文件读取失败: %w", err)
+	}
+
+	// 解析YAML配置
+	if err := yaml.Unmarshal(data, c); err != nil {
+		return fmt.Errorf("配置文件解析失败: %w", err)
+	}
+
+	return nil
+}
+
+// WriteTempFile 将数据写入到临时子目录的指定文件中。
+// 它会自动创建所需的子目录。
+// 注意：此函数本身不是线程安全的，应在 ExecuteWithLock 的回调中使用。
+func WriteTempFile(tempRoot, subDir, fileName string, data []byte) (string, error) {
+	fullDirPath := filepath.Join(tempRoot, subDir)
+	if err := os.MkdirAll(fullDirPath, 0755); err != nil {
+		return "", fmt.Errorf("为写入操作创建临时目录 %s 失败: %w", fullDirPath, err)
+	}
+
+	filePath := filepath.Join(fullDirPath, fileName)
+	if err := os.WriteFile(filePath, data, 0644); err != nil {
+		return "", fmt.Errorf("写入临时文件 %s 失败: %w", filePath, err)
+	}
+
+	return filePath, nil
+}
+
+// ReadTempFile 从临时子目录的指定文件中读取数据。
+// 注意：此函数本身不是线程安全的，应在 ExecuteWithLock 的回调中使用。
+func ReadTempFile(tempRoot, subDir, fileName string) ([]byte, error) {
+	filePath := filepath.Join(tempRoot, subDir, fileName)
+	data, err := os.ReadFile(filePath)
+	if err != nil {
+		return nil, fmt.Errorf("读取临时文件 %s 失败: %w", filePath, err)
+	}
+	return data, nil
+}