实现 StartUpgrade

internal/core/component_initializers.go

@@ -147,6 +147,7 @@ type DomainServices struct {
 	pigBatchDomain          pig.PigBatchService
 	deviceOperator          device.DeviceOperator
 	deviceCommunicator      device.DeviceCommunicator
+	otaService              device.OtaService
 	taskFactory             plan.TaskFactory
 	planExecutionManager    plan.ExecutionManager
 	analysisPlanTaskManager plan.AnalysisPlanTaskManager
@@ -191,6 +192,15 @@ func initDomainServices(ctx context.Context, cfg *config.Config, infra *Infrastr
 		infra.repos.pendingCollectionRepo,
 		infra.lora.comm,
 	)
+	otaService := device.NewOtaService(
+		logs.AddCompName(baseCtx, "OtaService"),
+		device.OtaConfig{
+			DefaultRetryCount:      uint32(cfg.OTA.DefaultRetryCount),
+			DefaultRequestTimeoutS: uint32(cfg.OTA.DefaultRequestTimeoutSeconds),
+		},
+		infra.repos.otaRepo,
+		generalDeviceService,
+	)
 
 	// 告警服务
 	alarmService := alarm.NewAlarmService(
@@ -268,6 +278,7 @@ func initDomainServices(ctx context.Context, cfg *config.Config, infra *Infrastr
 		pigBatchDomain:          pigBatchDomain,
 		deviceOperator:          generalDeviceService,
 		deviceCommunicator:      generalDeviceService,
+		otaService:              otaService,
 		analysisPlanTaskManager: analysisPlanTaskManager,
 		taskFactory:             taskFactory,
 		planExecutionManager:    planExecutionManager,
@@ -344,6 +355,7 @@ func initAppServices(ctx context.Context, infra *Infrastructure, domainServices
 		logs.AddCompName(baseCtx, "AreaControllerService"),
 		infra.repos.areaControllerRepo,
 		thresholdAlarmService,
+		domainServices.otaService,
 	)
 
 	auditService := service.NewAuditService(logs.AddCompName(baseCtx, "AuditService"), infra.repos.userActionLogRepo)

internal/domain/device/ota_service.go

@@ -8,6 +8,7 @@ import (
 	"crypto/sha256"
 	"crypto/x509"
 	"encoding/base64"
+	"encoding/json"
 	"encoding/pem"
 	"fmt"
 	"io/fs"
@@ -19,8 +20,8 @@ import (
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/file"
-
 	"github.com/gibson042/canonicaljson-go"
 )
 
@@ -56,25 +57,157 @@ type ManifestFile struct {
 	Size int64  `json:"size"` // 文件的大小（字节）
 }
 
+// OtaConfig 封装了 OTA 服务所需的可配置参数。
+type OtaConfig struct {
+	DefaultRetryCount      uint32 // 默认的设备端文件下载重试次数
+	DefaultRequestTimeoutS uint32 // 默认的设备端文件下载请求超时时间（秒）
+}
+
 // otaServiceImpl 是 OtaService 接口的实现。
 type otaServiceImpl struct {
-	ctx        context.Context
+	ctx                  context.Context
-	otaRepo    repository.OtaRepository
+	config               OtaConfig
-	deviceRepo repository.DeviceRepository
+	otaRepo              repository.OtaRepository
+	generalDeviceService *GeneralDeviceService
 }
 
 // NewOtaService 创建一个新的 OtaService 实例。
-func NewOtaService(ctx context.Context, otaRepo repository.OtaRepository, deviceRepo repository.DeviceRepository) OtaService {
+func NewOtaService(
+	ctx context.Context,
+	config OtaConfig,
+	otaRepo repository.OtaRepository,
+	generalDeviceService *GeneralDeviceService,
+) OtaService {
 	return &otaServiceImpl{
-		ctx:        ctx,
+		ctx:                  ctx,
-		otaRepo:    otaRepo,
+		config:               config,
-		deviceRepo: deviceRepo,
+		otaRepo:              otaRepo,
+		generalDeviceService: generalDeviceService,
+	}
+}
+
+// upgradeTask 封装了单次升级任务的所有上下文和操作，以提高代码的可读性和模块化。
+type upgradeTask struct {
+	service      *otaServiceImpl
+	ctx          context.Context
+	logger       *logs.Logger
+	task         *models.OTATask
+	firmwarePath string
+	tempSubDir   string
+}
+
+// run 执行核心的升级准备流程。
+// 此方法内的所有操作都处于一个文件锁的保护下。
+func (t *upgradeTask) run() error {
+	// 步骤 1: 解压固件
+	tempDestPath, err := file.CreateTempDir(t.tempSubDir)
+	if err != nil {
+		return fmt.Errorf("创建临时目录失败: %w", err)
+	}
+	if err := file.Decompress(t.firmwarePath, tempDestPath); err != nil {
+		return fmt.Errorf("解压固件失败: %w", err)
+	}
+	t.logger.Infof("为任务 %d 成功解压固件到 %s", t.task.ID, tempDestPath)
+
+	// 步骤 2: 生成、签名并写入 manifest 文件
+	manifest, err := t.service.generateManifest(t.tempSubDir)
+	if err != nil {
+		return fmt.Errorf("生成 manifest 失败: %w", err)
+	}
+	if err := t.service.signManifest(manifest); err != nil {
+		return fmt.Errorf("签名 manifest 失败: %w", err)
+	}
+	manifestBytes, err := json.Marshal(manifest)
+	if err != nil {
+		return fmt.Errorf("序列化 manifest.json 失败: %w", err)
+	}
+	if _, err := file.WriteTempFile(t.tempSubDir, "manifest.json", manifestBytes); err != nil {
+		return fmt.Errorf("写入 manifest.json 失败: %w", err)
+	}
+	t.logger.Infof("为任务 %d 成功生成并签名 manifest.json", t.task.ID)
+
+	// 步骤 3: 发送升级指令
+	manifestMD5 := fmt.Sprintf("%x", md5.Sum(manifestBytes))
+	prepareReq := &proto.PrepareUpdateReq{
+		Version:               manifest.Version,
+		TaskId:                t.task.ID,
+		ManifestMd5:           manifestMD5,
+		RetryCount:            t.service.config.DefaultRetryCount,
+		RequestTimeoutSeconds: t.service.config.DefaultRequestTimeoutS,
+	}
+	instructionPayload := &proto.Instruction_PrepareUpdateReq{PrepareUpdateReq: prepareReq}
+	if err := t.service.generalDeviceService.Send(t.ctx, t.task.AreaControllerID, instructionPayload); err != nil {
+		return fmt.Errorf("发送升级指令失败: %w", err)
+	}
+	t.logger.Infof("为任务 %d 成功发送升级指令", t.task.ID)
+
+	// 步骤 4: 更新任务状态为“进行中”
+	t.task.Status = models.OTATaskStatusInProgress
+	t.task.TargetVersion = manifest.Version // 回填目标版本号
+	if err := t.service.otaRepo.Update(t.ctx, t.task); err != nil {
+		return fmt.Errorf("更新任务状态为 '进行中' 失败: %w", err)
+	}
+
+	return nil
+}
+
+// rollback 在 run 方法失败时执行清理和状态更新操作。
+func (t *upgradeTask) rollback(originalErr error) {
+	t.logger.Errorf("任务 %d 文件准备阶段失败，执行回滚: %v", t.task.ID, originalErr)
+
+	// 更新数据库状态为“准备文件失败”
+	t.task.Status = models.OTATaskStatusFailedPreparation
+	t.task.ErrorMessage = fmt.Sprintf("文件准备阶段失败: %v", originalErr)
+	now := time.Now()
+	t.task.CompletedAt = &now
+	if updateErr := t.service.otaRepo.Update(t.ctx, t.task); updateErr != nil {
+		t.logger.DPanicf("CRITICAL: 任务 %d 回滚失败后，更新其状态也失败了: %v", t.task.ID, updateErr)
+	}
+
+	// 清理临时解压目录
+	if removeDirErr := file.RemoveTempDir(t.tempSubDir); removeDirErr != nil {
+		t.logger.Warnf("回滚操作：清理任务 %d 的临时目录 %s 失败: %v", t.task.ID, t.tempSubDir, removeDirErr)
+	}
+	// 清理原始固件压缩包
+	if removeSrcErr := os.Remove(t.firmwarePath); removeSrcErr != nil {
+		t.logger.Warnf("回滚操作：清理任务 %d 的源固件 %s 失败: %v", t.task.ID, t.firmwarePath, removeSrcErr)
 	}
 }
 
 func (o *otaServiceImpl) StartUpgrade(ctx context.Context, areaControllerID uint32, firmwarePath string) (uint32, error) {
-	//TODO implement me
+	serviceCtx, logger := logs.Trace(ctx, o.ctx, "StartUpgrade")
-	panic("implement me")
+
+	// 步骤 1: 预创建数据库记录
+	task := &models.OTATask{
+		AreaControllerID: areaControllerID,
+		Status:           models.OTATaskStatusPending,
+		CreatedAt:        time.Now(),
+	}
+	if err := o.otaRepo.Create(serviceCtx, task); err != nil {
+		logger.Errorf("预创建 OTA 任务记录失败: %v", err)
+		return 0, fmt.Errorf("预创建 OTA 任务记录失败: %w", err)
+	}
+	logger.Infof("成功预创建 OTA 任务记录, ID: %d", task.ID)
+
+	// 步骤 2: 初始化升级任务执行器
+	upgrade := &upgradeTask{
+		service:      o,
+		ctx:          serviceCtx,
+		logger:       logger,
+		task:         task,
+		firmwarePath: firmwarePath,
+		tempSubDir:   filepath.Join(models.OTADir, fmt.Sprintf("%d", task.ID)),
+	}
+
+	// 步骤 3: 在文件锁的保护下，原子化地执行升级准备流程
+	if err := file.ExecuteWithLock(upgrade.run, upgrade.rollback); err != nil {
+		// 此处的错误已在 rollback 中处理和记录，这里只向调用方返回失败信号
+		logger.Errorf("OTA 任务 %d 未能成功启动: %v", task.ID, err)
+		return 0, err
+	}
+
+	logger.Infof("OTA 升级任务 %d 已成功启动", task.ID)
+	return task.ID, nil
 }
 
 func (o *otaServiceImpl) GetUpgradeProgress(ctx context.Context, taskID uint32) (executed, total uint32, CurrentStage models.OTATaskStatus, err error) {
@@ -134,7 +267,7 @@ func (o *otaServiceImpl) generateManifest(packageSubDir string) (*Manifest, erro
 		if err != nil {
 			return err
 		}
-		if d.Name() == "version" {
+		if d.Name() == "version" || d.Name() == "manifest.json" {
 			return nil
 		}
 

internal/infra/models/execution.go

@@ -166,12 +166,13 @@ const (
 	OTATaskStatusSuccess         OTATaskStatus = "成功"     // 设备报告升级成功，新固件已运行
 	OTATaskStatusAlreadyUpToDate OTATaskStatus = "版本已是最新" // 设备报告版本已是最新，未执行升级
 
-	OTATaskStatusFailedPreCheck OTATaskStatus = "预检失败"    // 设备报告升级前检查失败 (如拒绝降级、准备分区失败)
+	OTATaskStatusFailedPreparation OTATaskStatus = "准备升级失败"  // 平台在解压、生成清单等文件操作阶段发生错误
-	OTATaskStatusFailedDownload OTATaskStatus = "下载或校验失败" // 设备报告文件下载或校验失败 (包括清单文件和固件文件)
+	OTATaskStatusFailedPreCheck    OTATaskStatus = "预检失败"    // 设备报告升级前检查失败 (如拒绝降级、准备分区失败)
-	OTATaskStatusFailedRollback OTATaskStatus = "固件回滚"    // 新固件启动失败，设备自动回滚
+	OTATaskStatusFailedDownload    OTATaskStatus = "下载或校验失败" // 设备报告文件下载或校验失败 (包括清单文件和固件文件)
-	OTATaskStatusTimedOut       OTATaskStatus = "超时"      // 平台在超时后仍未收到最终报告
+	OTATaskStatusFailedRollback    OTATaskStatus = "固件回滚"    // 新固件启动失败，设备自动回滚
-	OTATaskStatusPlatformError  OTATaskStatus = "平台内部错误"  // 平台处理过程中发生的非设备报告错误
+	OTATaskStatusTimedOut          OTATaskStatus = "超时"      // 平台在超时后仍未收到最终报告
-	OTATaskStatusStopped        OTATaskStatus = "手动停止"    // 手动停止
+	OTATaskStatusPlatformError     OTATaskStatus = "平台内部错误"  // 平台处理过程中发生的非设备报告错误
+	OTATaskStatusStopped           OTATaskStatus = "手动停止"    // 手动停止
 )
 
 // OTADir 是 OTA 升级相关的临时文件存储目录

internal/infra/repository/ota_repository.go

@@ -14,12 +14,16 @@ import (
 type OtaRepository interface {
 	// Create 创建一个新的 OTA 任务。
 	Create(ctx context.Context, task *models.OTATask) error
+	// CreateTx 在指定的事务中创建一个新的 OTA 任务。
+	CreateTx(ctx context.Context, tx *gorm.DB, 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 任务。
 	Update(ctx context.Context, task *models.OTATask) error
+	// UpdateTx 在指定的事务中更新单个 OTA 任务。
+	UpdateTx(ctx context.Context, tx *gorm.DB, task *models.OTATask) error
 }
 
 // gormOtaRepository 是 OtaRepository 的 GORM 实现
@@ -36,10 +40,17 @@ func NewGormOtaRepository(ctx context.Context, db *gorm.DB) OtaRepository {
 	}
 }
 
-// Create 实现了创建新 OTA 任务的逻辑。
+// Create 实现了创建新 OTA 任务的逻辑，内部调用 CreateTx 以复用代码。
 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
+	// 使用 r.db 作为事务对象，调用通用的事务方法
+	return r.CreateTx(repoCtx, r.db, task)
+}
+
+// CreateTx 实现了在事务中创建新 OTA 任务的核心逻辑。
+func (r *gormOtaRepository) CreateTx(ctx context.Context, tx *gorm.DB, task *models.OTATask) error {
+	repoCtx := logs.AddFuncName(ctx, r.ctx, "CreateTx")
+	return tx.WithContext(repoCtx).Create(task).Error
 }
 
 // FindByID 实现了根据 ID 查找任务的逻辑。
@@ -47,7 +58,10 @@ func (r *gormOtaRepository) FindByID(ctx context.Context, id uint32) (*models.OT
 	repoCtx := logs.AddFuncName(ctx, r.ctx, "FindByID")
 	var task models.OTATask
 	err := r.db.WithContext(repoCtx).First(&task, id).Error
-	return &task, err
+	if err != nil {
+		return nil, err
+	}
+	return &task, nil
 }
 
 // FindTasksByStatusesAndCreationTime 实现了根据状态和创建时间查找任务的逻辑。
@@ -63,8 +77,15 @@ func (r *gormOtaRepository) FindTasksByStatusesAndCreationTime(ctx context.Conte
 	return tasks, err
 }
 
-// Update 实现了更新单个 OTA 任务的逻辑。
+// Update 实现了更新单个 OTA 任务的逻辑，内部调用 UpdateTx 以复用代码。
 func (r *gormOtaRepository) Update(ctx context.Context, task *models.OTATask) error {
 	repoCtx := logs.AddFuncName(ctx, r.ctx, "Update")
-	return r.db.WithContext(repoCtx).Save(task).Error
+	// 使用 r.db 作为事务对象，调用通用的事务方法
+	return r.UpdateTx(repoCtx, r.db, task)
+}
+
+// UpdateTx 实现了在事务中更新单个 OTA 任务的核心逻辑。
+func (r *gormOtaRepository) UpdateTx(ctx context.Context, tx *gorm.DB, task *models.OTATask) error {
+	repoCtx := logs.AddFuncName(ctx, r.ctx, "UpdateTx")
+	return tx.WithContext(repoCtx).Save(task).Error
 }

internal/infra/utils/file/decompress.go

@@ -45,7 +45,7 @@ func (ct CompressionType) Matches(filename string) bool {
 // 此函数假定它拥有对 destPath 的完全控制权，并会在失败时将其彻底删除。
 func DecompressAtomic(sourcePath, destPath string) error {
 	action := func() error {
-		return decompress(sourcePath, destPath)
+		return Decompress(sourcePath, destPath)
 	}
 
 	onRollback := func(err error) {
@@ -55,9 +55,9 @@ func DecompressAtomic(sourcePath, destPath string) error {
 	return ExecuteWithLock(action, onRollback)
 }
 
-// decompress 是解压操作的核心实现，它本身不是线程安全的，也不提供回滚。
+// Decompress 是解压操作的核心实现，它本身不是线程安全的，也不提供回滚。
 // 它应该被 DecompressAtomic 或其他调用方在 ExecuteWithLock 的回调中执行。
-func decompress(sourcePath, destPath string) error {
+func Decompress(sourcePath, destPath string) error {
 	// 确保目标目录存在
 	if err := os.MkdirAll(destPath, 0755); err != nil {
 		return fmt.Errorf("创建目标目录 %s 失败: %w", destPath, err)