增加ping指令并获取带版本号的响应

internal/domain/task/heartbeat_task.go

@@ -0,0 +1,93 @@
+package task
+
+import (
+	"context"
+	"fmt"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/plan"
+	"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"
+)
+
+// HeartbeatTask 实现了 plan.Task 接口，用于执行一次区域主控心跳检测（发送Ping）
+type HeartbeatTask struct {
+	ctx                context.Context
+	log                *models.TaskExecutionLog
+	areaControllerRepo repository.AreaControllerRepository
+	deviceService      device.Service
+}
+
+// NewHeartbeatTask 创建一个心跳检测任务实例
+func NewHeartbeatTask(
+	ctx context.Context,
+	log *models.TaskExecutionLog,
+	areaControllerRepo repository.AreaControllerRepository,
+	deviceService device.Service,
+) plan.Task {
+	return &HeartbeatTask{
+		ctx:                ctx,
+		log:                log,
+		areaControllerRepo: areaControllerRepo,
+		deviceService:      deviceService,
+	}
+}
+
+// Execute 是任务的核心执行逻辑
+func (t *HeartbeatTask) Execute(ctx context.Context) error {
+	taskCtx, logger := logs.Trace(ctx, t.ctx, "Execute")
+	logger.Infow("开始执行区域主控心跳检测任务", "task_id", t.log.TaskID, "task_type", t.log.Task.Type, "log_id", t.log.ID)
+
+	controllers, err := t.areaControllerRepo.ListAll(taskCtx)
+	if err != nil {
+		return fmt.Errorf("心跳检测任务：获取所有区域主控失败: %w", err)
+	}
+
+	if len(controllers) == 0 {
+		logger.Infow("心跳检测任务：未发现任何区域主控，跳过本次检测")
+		return nil
+	}
+
+	// 构建 Ping 指令
+	pingInstruction := &proto.Instruction_Ping{
+		Ping: &proto.Ping{},
+	}
+
+	var firstError error
+	for _, controller := range controllers {
+		logger.Infow("向区域主控发送Ping指令", "controller_id", controller.ID)
+		err := t.deviceService.Send(taskCtx, controller.ID, pingInstruction, device.WithoutTracking())
+		if err != nil {
+			logger.Errorw("向区域主控发送Ping指令失败", "controller_id", controller.ID, "error", err)
+			if firstError == nil {
+				firstError = err // 保存第一个发生的错误
+			}
+		}
+	}
+
+	if firstError != nil {
+		return fmt.Errorf("心跳检测任务执行期间发生错误: %w", firstError)
+	}
+
+	logger.Infow("区域主控心跳检测任务执行完成", "task_id", t.log.TaskID, "task_type", t.log.Task.Type, "log_id", t.log.ID)
+	return nil
+}
+
+// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑
+func (t *HeartbeatTask) OnFailure(ctx context.Context, executeErr error) {
+	logger := logs.TraceLogger(ctx, t.ctx, "OnFailure")
+	logger.Errorw("区域主控心跳检测任务执行失败",
+		"task_id", t.log.TaskID,
+		"task_type", t.log.Task.Type,
+		"log_id", t.log.ID,
+		"error", executeErr,
+	)
+}
+
+// ResolveDeviceIDs 获取当前任务需要使用的设备ID列表
+func (t *HeartbeatTask) ResolveDeviceIDs(ctx context.Context) ([]uint32, error) {
+	// 心跳检测任务不和任何特定设备绑定
+	return []uint32{}, nil
+}

internal/domain/task/task.go

@@ -18,14 +18,16 @@ const (
 	CompNameReleaseFeedWeight  = "ReleaseFeedWeightTask"
 	CompNameFullCollectionTask = "FullCollectionTask"
 	CompNameAlarmNotification  = "AlarmNotificationTask"
+	CompNameHeartbeatTask      = "HeartbeatTask"
 )
 
 type taskFactory struct {
 	ctx context.Context
 
-	sensorDataRepo repository.SensorDataRepository
-	deviceRepo     repository.DeviceRepository
-	alarmRepo      repository.AlarmRepository
+	sensorDataRepo     repository.SensorDataRepository
+	deviceRepo         repository.DeviceRepository
+	alarmRepo          repository.AlarmRepository
+	areaControllerRepo repository.AreaControllerRepository
 
 	deviceService       device.Service
 	notificationService notify.Service
@@ -37,6 +39,7 @@ func NewTaskFactory(
 	sensorDataRepo repository.SensorDataRepository,
 	deviceRepo repository.DeviceRepository,
 	alarmRepo repository.AlarmRepository,
+	areaControllerRepo repository.AreaControllerRepository,
 	deviceService device.Service,
 	notifyService notify.Service,
 	alarmService alarm.AlarmService,
@@ -46,6 +49,7 @@ func NewTaskFactory(
 		sensorDataRepo:      sensorDataRepo,
 		deviceRepo:          deviceRepo,
 		alarmRepo:           alarmRepo,
+		areaControllerRepo:  areaControllerRepo,
 		deviceService:       deviceService,
 		notificationService: notifyService,
 		alarmService:        alarmService,
@@ -62,6 +66,8 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
 		return NewReleaseFeedWeightTask(logs.AddCompName(baseCtx, CompNameReleaseFeedWeight), claimedLog, t.sensorDataRepo, t.deviceRepo, t.deviceService)
 	case models.TaskTypeFullCollection:
 		return NewFullCollectionTask(logs.AddCompName(baseCtx, CompNameFullCollectionTask), claimedLog, t.deviceRepo, t.deviceService)
+	case models.TaskTypeHeartbeat:
+		return NewHeartbeatTask(logs.AddCompName(baseCtx, CompNameHeartbeatTask), claimedLog, t.areaControllerRepo, t.deviceService)
 	case models.TaskTypeAlarmNotification:
 		return NewAlarmNotificationTask(logs.AddCompName(baseCtx, CompNameAlarmNotification), claimedLog, t.notificationService, t.alarmRepo)
 	case models.TaskTypeDeviceThresholdCheck:
@@ -71,7 +77,6 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
 	case models.TaskTypeNotificationRefresh:
 		return NewRefreshNotificationTask(logs.AddCompName(baseCtx, "NotificationRefreshTask"), claimedLog, t.alarmService)
 	default:
-		// TODO 这里直接panic合适吗? 不过这个场景确实不该出现任何异常的任务类型
 		logger.Panicf("不支持的任务类型: %s", claimedLog.Task.Type)
 		panic("不支持的任务类型") // 显式panic防编译器报错
 	}
@@ -79,8 +84,6 @@ func (t *taskFactory) Production(ctx context.Context, claimedLog *models.TaskExe
 
 // CreateTaskFromModel 实现了 TaskFactory 接口，用于从模型创建任务实例。
 func (t *taskFactory) CreateTaskFromModel(ctx context.Context, taskModel *models.Task) (plan.TaskDeviceIDResolver, error) {
-	// 这个方法不关心 claimedLog 的其他字段，所以可以构造一个临时的
-	// 它只用于访问那些不依赖于执行日志的方法，比如 ResolveDeviceIDs
 	tempLog := &models.TaskExecutionLog{Task: *taskModel}
 	baseCtx := context.Background()
 
@@ -97,6 +100,8 @@ func (t *taskFactory) CreateTaskFromModel(ctx context.Context, taskModel *models
 		), nil
 	case models.TaskTypeFullCollection:
 		return NewFullCollectionTask(logs.AddCompName(baseCtx, CompNameFullCollectionTask), tempLog, t.deviceRepo, t.deviceService), nil
+	case models.TaskTypeHeartbeat:
+		return NewHeartbeatTask(logs.AddCompName(baseCtx, CompNameHeartbeatTask), tempLog, t.areaControllerRepo, t.deviceService), nil
 	case models.TaskTypeAlarmNotification:
 		return NewAlarmNotificationTask(logs.AddCompName(baseCtx, CompNameAlarmNotification), tempLog, t.notificationService, t.alarmRepo), nil
 	case models.TaskTypeDeviceThresholdCheck: