提供lora公共逻辑

internal/app/listener/chirp_stack/chirp_stack.go

@@ -4,19 +4,16 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"io"
-	"math"
 	"net/http"
-	"time"
 
 	"git.huangwc.com/pig/pig-farm-controller/internal/app/listener"
 	"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"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
 
 	gproto "google.golang.org/protobuf/proto"
-	"gorm.io/datatypes"
 )
 
 // ChirpStackListener 主动发送的请求的event字段, 这个字段代表事件类型
@@ -31,45 +28,34 @@ const (
 	eventTypeIntegration = "integration" // 集成事件：当其他集成（如第三方服务）处理数据后触发。
 )
 
-// ChirpStackListener 是一个监听器, 用于监听ChirpStack反馈的设备上行事件
+// ChirpStackListener 是一个监听器, 用于将ChirpStack的Webhook事件适配到统一的UpstreamHandler。
 type ChirpStackListener struct {
-	ctx                   context.Context
-	sensorDataRepo        repository.SensorDataRepository
-	deviceRepo            repository.DeviceRepository
-	areaControllerRepo    repository.AreaControllerRepository
-	deviceCommandLogRepo  repository.DeviceCommandLogRepository
-	pendingCollectionRepo repository.PendingCollectionRepository
+	selfCtx context.Context
+	handler transport.UpstreamHandler // 依赖注入的统一业务处理器
 }
 
-// NewChirpStackListener 创建一个新的 ChirpStackListener 实例
+// NewChirpStackListener 创建一个新的 ChirpStackListener 实例。
 func NewChirpStackListener(
 	ctx context.Context,
-	sensorDataRepo repository.SensorDataRepository,
-	deviceRepo repository.DeviceRepository,
-	areaControllerRepo repository.AreaControllerRepository,
-	deviceCommandLogRepo repository.DeviceCommandLogRepository,
-	pendingCollectionRepo repository.PendingCollectionRepository,
+	handler transport.UpstreamHandler,
 ) listener.ListenHandler {
 	return &ChirpStackListener{
-		ctx:                   ctx,
-		sensorDataRepo:        sensorDataRepo,
-		deviceRepo:            deviceRepo,
-		areaControllerRepo:    areaControllerRepo,
-		deviceCommandLogRepo:  deviceCommandLogRepo,
-		pendingCollectionRepo: pendingCollectionRepo,
+		selfCtx: logs.AddCompName(ctx, "ChirpStackListener"),
+		handler: handler,
 	}
 }
 
 // Handler 监听ChirpStack反馈的事件, 因为这是个Webhook, 所以直接回复掉再慢慢处理信息
 func (c *ChirpStackListener) Handler() http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		ctx, logger := logs.Trace(r.Context(), c.ctx, "ChirpStackListener")
+		// 注意：这里的 selfCtx 是 r.Context()，因为它包含了HTTP请求的追踪信息
+		ctx, logger := logs.Trace(r.Context(), c.selfCtx, "Handler")
 
 		defer r.Body.Close()
 
 		b, err := io.ReadAll(r.Body)
 		if err != nil {
-			logger.Errorf("读取请求体失败: %v", err)
+			logger.Errorw("读取请求体失败", "error", err)
 			http.Error(w, "failed to read body", http.StatusBadRequest)
 			return
 		}
@@ -78,379 +64,112 @@ func (c *ChirpStackListener) Handler() http.HandlerFunc {
 
 		w.WriteHeader(http.StatusOK)
 
-		// 将异步处理逻辑委托给 handler 方法
-		go c.handler(ctx, b, event)
+		// 使用分离的上下文进行异步处理，防止原始请求取消导致处理中断
+		detachedCtx := logs.DetachContext(ctx)
+		go c.dispatch(detachedCtx, b, event)
 	}
 }
 
-// handler 用于处理 ChirpStack 发送的事件
-func (c *ChirpStackListener) handler(ctx context.Context, data []byte, eventType string) {
-	reqCtx, logger := logs.Trace(ctx, c.ctx, "ChirpStackListener.handler")
+// dispatch 用于解析并分发 ChirpStack 发送的事件
+func (c *ChirpStackListener) dispatch(ctx context.Context, data []byte, eventType string) {
+	reqCtx, logger := logs.Trace(ctx, c.selfCtx, "dispatch")
+
+	var err error
 	switch eventType {
 	case eventTypeUp:
 		var msg UpEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'up' 事件失败: %v, data: %s", err, string(data))
-			return
+		if err = json.Unmarshal(data, &msg); err == nil {
+			c.adaptUpEvent(reqCtx, &msg)
 		}
-		c.handleUpEvent(reqCtx, &msg)
-
-	case eventTypeJoin:
-		var msg JoinEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'join' 事件失败: %v, data: %s", err, string(data))
-			return
-		}
-		c.handleJoinEvent(reqCtx, &msg)
-
-	case eventTypeAck:
-		var msg AckEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'ack' 事件失败: %v, data: %s", err, string(data))
-			return
-		}
-		c.handleAckEvent(reqCtx, &msg)
-
-	case eventTypeTxAck:
-		var msg TxAckEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'txack' 事件失败: %v, data: %s", err, string(data))
-			return
-		}
-		c.handleTxAckEvent(reqCtx, &msg)
 
 	case eventTypeStatus:
 		var msg StatusEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'status' 事件失败: %v, data: %s", err, string(data))
-			return
+		if err = json.Unmarshal(data, &msg); err == nil {
+			c.adaptStatusEvent(reqCtx, &msg)
 		}
-		c.handleStatusEvent(reqCtx, &msg)
 
-	case eventTypeLog:
-		var msg LogEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'log' 事件失败: %v, data: %s", err, string(data))
-			return
+	case eventTypeAck:
+		var msg AckEvent
+		if err = json.Unmarshal(data, &msg); err == nil {
+			c.adaptAckEvent(reqCtx, &msg)
 		}
-		c.handleLogEvent(reqCtx, &msg)
 
-	case eventTypeLocation:
-		var msg LocationEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'location' 事件失败: %v, data: %s", err, string(data))
-			return
-		}
-		c.handleLocationEvent(reqCtx, &msg)
-
-	case eventTypeIntegration:
-		var msg IntegrationEvent
-		if err := json.Unmarshal(data, &msg); err != nil {
-			logger.Errorf("解析 'integration' 事件失败: %v, data: %s", err, string(data))
-			return
-		}
-		c.handleIntegrationEvent(reqCtx, &msg)
+	// --- 其他事件只记录日志，不进行业务处理 ---
+	case eventTypeJoin, eventTypeTxAck, eventTypeLog, eventTypeLocation, eventTypeIntegration:
+		logger.Infow("收到一个非业务处理的ChirpStack事件", "type", eventType)
 
 	default:
-		logger.Errorf("未知的ChirpStack事件: %s, data: %s", eventType, string(data))
+		logger.Warnw("未知的ChirpStack事件类型", "type", eventType)
+	}
+
+	if err != nil {
+		logger.Errorw("解析ChirpStack事件失败", "type", eventType, "error", err, "data", string(data))
 	}
 }
 
-// --- 业务处理函数 ---
+// --- 适配器函数 ---
 
-// handleUpEvent 处理上行数据事件
-func (c *ChirpStackListener) handleUpEvent(ctx context.Context, event *UpEvent) {
-	reqCtx, logger := logs.Trace(ctx, c.ctx, "ChirpStackListener.handleUpEvent")
-	logger.Infof("开始处理 'up' 事件, DevEui: %s", event.DeviceInfo.DevEui)
+// adaptUpEvent 将 'up' 事件适配并委托给 UpstreamHandler
+func (c *ChirpStackListener) adaptUpEvent(ctx context.Context, event *UpEvent) {
+	reqCtx, logger := logs.Trace(ctx, c.selfCtx, "adaptUpEvent")
 
-	// 1. 查找区域主控设备
-	areaController, err := c.areaControllerRepo.FindByNetworkID(reqCtx, event.DeviceInfo.DevEui)
-	if err != nil {
-		logger.Errorf("处理 'up' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err)
-		return
-	}
-	// 依赖 SelfCheck 确保区域主控有效
-	if err := areaController.SelfCheck(); err != nil {
-		logger.Errorf("处理 'up' 事件失败：区域主控 %v(ID: %d) 未通过自检: %v", areaController.Name, areaController.ID, err)
-		return
-	}
-	logger.Infof("找到区域主控: %s (ID: %d)", areaController.Name, areaController.ID)
-
-	// 2. 记录区域主控的信号强度 (如果存在)
+	// 1. 优先处理并委托旁路状态信息（如信号强度）
 	if len(event.RxInfo) > 0 {
-		// 根据业务逻辑，一个猪场只有一个网关，所以 RxInfo 中通常只有一个元素，或者 gateway_id 都是相同的。
-		// 因此，我们只取第一个 RxInfo 中的信号数据即可。
-		rx := event.RxInfo[0] // 取第一个接收到的网关信息
-
-		// 构建 SignalMetrics 结构体
-		signalMetrics := models.SignalMetrics{
-			RssiDbm: rx.Rssi,
-			SnrDb:   rx.Snr,
+		rx := event.RxInfo[0]
+		status := map[string]interface{}{
+			"rssi": float64(rx.Rssi),
+			"snr":  float64(rx.Snr),
+		}
+		if err := c.handler.HandleStatus(reqCtx, event.DeviceInfo.DevEui, status); err != nil {
+			logger.Errorw("委托 'up' 事件中的状态信息失败", "error", err)
 		}
-
-		// 记录信号强度
-		c.recordSensorData(reqCtx, areaController.ID, areaController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics)
-		logger.Infof("已记录区域主控 (ID: %d) 的信号强度: RSSI=%d, SNR=%.2f", areaController.ID, rx.Rssi, rx.Snr)
-	} else {
-		logger.Warnf("处理 'up' 事件时未找到 RxInfo，无法记录信号数据。DevEui: %s", event.DeviceInfo.DevEui)
 	}
 
-	// 3. 处理上报的传感器数据
+	// 2. 如果没有业务数据，则直接返回
 	if event.Data == "" {
-		logger.Warnf("处理 'up' 事件时 Data 字段为空，无需记录上行数据。DevEui: %s", event.DeviceInfo.DevEui)
 		return
 	}
 
-	// 3.1 Base64 解码
+	// 3. 解码并解析业务指令
 	decodedData, err := base64.StdEncoding.DecodeString(event.Data)
 	if err != nil {
-		logger.Errorf("Base64 解码 'up' 事件的 Data 失败: %v, Data: %s", err, event.Data)
+		logger.Errorw("Base64解码 'up' 事件的Data失败", "error", err, "data", event.Data)
 		return
 	}
 
-	// 3.2 解析外层 "信封"
 	var instruction proto.Instruction
 	if err := gproto.Unmarshal(decodedData, &instruction); err != nil {
-		logger.Errorf("解析上行 Instruction Protobuf 失败: %v, Decoded Data: %x", err, decodedData)
+		logger.Errorw("解析上行Instruction Protobuf失败", "error", err, "decodedData", fmt.Sprintf("%x", decodedData))
 		return
 	}
 
-	// 3.3 使用 type switch 从 oneof payload 中提取 CollectResult
-	var collectResp *proto.CollectResult
-	switch p := instruction.GetPayload().(type) {
-	case *proto.Instruction_CollectResult:
-		collectResp = p.CollectResult
-	default:
-		// 如果上行的数据不是采集结果，记录日志并忽略
-		logger.Infof("收到一个非采集响应的上行指令 (Type: %T)，无需处理。", p)
-		return
-	}
-
-	// 检查 collectResp 是否为 nil，虽然在 type switch 成功的情况下不太可能
-	if collectResp == nil {
-		logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil")
-		return
-	}
-
-	correlationID := collectResp.CorrelationId
-	logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values))
-
-	// 4. 根据 CorrelationID 查找待处理请求
-	pendingReq, err := c.pendingCollectionRepo.FindByCorrelationID(reqCtx, correlationID)
-	if err != nil {
-		logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err)
-		return
-	}
-
-	// 检查状态，防止重复处理
-	if pendingReq.Status != models.PendingStatusPending && pendingReq.Status != models.PendingStatusTimedOut {
-		logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status)
-		return
-	}
-
-	// 5. 匹配数据并存入数据库
-	deviceIDs := pendingReq.CommandMetadata
-	values := collectResp.Values
-	if len(deviceIDs) != len(values) {
-		logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID)
-		// 即使数量不匹配，也更新状态为完成，以防止请求永远 pending
-		err = c.pendingCollectionRepo.UpdateStatusToFulfilled(reqCtx, correlationID, event.Time)
-		if err != nil {
-			logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err)
-		}
-		return
-	}
-
-	for i, deviceID := range deviceIDs {
-		rawSensorValue := values[i] // 这是设备上报的原始值
-
-		// 检查设备上报的值是否为 NaN (Not a Number)，如果是则跳过
-		if math.IsNaN(float64(rawSensorValue)) {
-			logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID)
-			continue
-		}
-
-		// 5.1 获取设备及其模板
-		dev, err := c.deviceRepo.FindByID(reqCtx, deviceID)
-		if err != nil {
-			logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err)
-			continue
-		}
-		// 依赖 SelfCheck 确保设备和模板有效
-		if err := dev.SelfCheck(); err != nil {
-			logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err)
-			continue
-		}
-		if err := dev.DeviceTemplate.SelfCheck(); err != nil {
-			logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err)
-			continue
-		}
-
-		// 5.2 从设备模板中解析 ValueDescriptor
-		var valueDescriptors []*models.ValueDescriptor
-		if err := dev.DeviceTemplate.ParseValues(&valueDescriptors); err != nil {
-			logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err)
-			continue
-		}
-		// 根据 DeviceTemplate.SelfCheck，这里应该只有一个 ValueDescriptor
-		if len(valueDescriptors) == 0 {
-			logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID)
-			continue
-		}
-		valueDescriptor := valueDescriptors[0]
-
-		// 5.3 应用乘数和偏移量计算最终值
-		parsedValue := rawSensorValue*valueDescriptor.Multiplier + valueDescriptor.Offset
-
-		// 5.4 根据传感器类型构建具体的数据结构
-		var dataToRecord interface{}
-		switch valueDescriptor.Type {
-		case models.SensorTypeTemperature:
-			dataToRecord = models.TemperatureData{TemperatureCelsius: parsedValue}
-		case models.SensorTypeHumidity:
-			dataToRecord = models.HumidityData{HumidityPercent: parsedValue}
-		case models.SensorTypeWeight:
-			dataToRecord = models.WeightData{WeightKilograms: parsedValue}
-		default:
-			// TODO 未知传感器的数据需要记录吗
-			logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type)
-			dataToRecord = map[string]float32{"value": parsedValue}
-		}
-
-		// 5.5 记录传感器数据
-		c.recordSensorData(reqCtx, areaController.ID, dev.ID, event.Time, valueDescriptor.Type, dataToRecord)
-		logger.Infof("成功记录传感器数据: 设备ID=%d, 类型=%s, 原始值=%f, 解析值=%.2f", dev.ID, valueDescriptor.Type, rawSensorValue, parsedValue)
-	}
-
-	// 6. 更新请求状态为“已完成”
-	if err := c.pendingCollectionRepo.UpdateStatusToFulfilled(reqCtx, correlationID, event.Time); err != nil {
-		logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err)
-	} else {
-		logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID)
+	// 4. 委托给统一处理器
+	if err := c.handler.HandleInstruction(reqCtx, event.DeviceInfo.DevEui, &instruction); err != nil {
+		logger.Errorw("委托 'up' 事件中的业务指令失败", "error", err)
 	}
 }
 
-// handleStatusEvent 处理设备状态事件
-func (c *ChirpStackListener) handleStatusEvent(ctx context.Context, event *StatusEvent) {
-	reqCtx, logger := logs.Trace(ctx, c.ctx, "handleStatusEvent")
-	logger.Infof("处接收到理 'status' 事件: %+v", event)
+// adaptStatusEvent 将 'status' 事件适配并委托给 UpstreamHandler
+func (c *ChirpStackListener) adaptStatusEvent(ctx context.Context, event *StatusEvent) {
+	reqCtx, logger := logs.Trace(ctx, c.selfCtx, "adaptStatusEvent")
 
-	// 查找区域主控设备
-	areaController, err := c.areaControllerRepo.FindByNetworkID(reqCtx, event.DeviceInfo.DevEui)
-	if err != nil {
-		logger.Errorf("处理 'status' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err)
-		return
+	status := map[string]interface{}{
+		"margin":                  float64(event.Margin),
+		"batteryLevel":            float64(event.BatteryLevel),
+		"batteryLevelUnavailable": event.BatteryLevelUnavailable,
+		"externalPower":           event.ExternalPower,
 	}
 
-	// 记录信号强度
-	signalMetrics := models.SignalMetrics{
-		MarginDb: event.Margin,
-	}
-	c.recordSensorData(reqCtx, areaController.ID, areaController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics)
-	logger.Infof("已记录区域主控 (ID: %d) 的信号状态: %+v", areaController.ID, signalMetrics)
-
-	// 记录电量
-	batteryLevel := models.BatteryLevel{
-		BatteryLevelRatio:       event.BatteryLevel,
-		BatteryLevelUnavailable: event.BatteryLevelUnavailable,
-		ExternalPower:           event.ExternalPower,
-	}
-	c.recordSensorData(reqCtx, areaController.ID, areaController.ID, event.Time, models.SensorTypeBatteryLevel, batteryLevel)
-	logger.Infof("已记录区域主控 (ID: %d) 的电池状态: %+v", areaController.ID, batteryLevel)
-}
-
-// handleAckEvent 处理下行确认事件
-func (c *ChirpStackListener) handleAckEvent(ctx context.Context, event *AckEvent) {
-	reqCtx, logger := logs.Trace(ctx, c.ctx, "handleAckEvent")
-	logger.Infof("接收到 'ack' 事件: %+v", event)
-
-	// 更新下行任务记录的确认时间及接收成功状态
-	err := c.deviceCommandLogRepo.UpdateAcknowledgedAt(reqCtx, event.DeduplicationID, event.Time, event.Acknowledged)
-	if err != nil {
-		logger.Errorf("更新下行任务记录的确认时间及接收成功状态失败 (MessageID: %s, DevEui: %s, Acknowledged: %t): %v",
-			event.DeduplicationID, event.DeviceInfo.DevEui, event.Acknowledged, err)
-		return
-	}
-
-	logger.Infof("成功更新下行任务记录确认时间及接收成功状态 (MessageID: %s, DevEui: %s, Acknowledged: %t, AcknowledgedAt: %s)",
-		event.DeduplicationID, event.DeviceInfo.DevEui, event.Acknowledged, event.Time.Format(time.RFC3339))
-}
-
-// handleLogEvent 处理日志事件
-func (c *ChirpStackListener) handleLogEvent(ctx context.Context, event *LogEvent) {
-	logger := logs.TraceLogger(ctx, c.ctx, "handleLogEvent")
-	// 首先，打印完整的事件结构体，用于详细排查
-	logger.Infof("接收到 'log' 事件的完整内容: %+v", event)
-
-	// 接着，根据 ChirpStack 日志的级别，使用我们自己的 logger 对应级别来打印核心信息
-	logMessage := "ChirpStack 日志: [%s] %s (DevEui: %s)"
-	switch event.Level {
-	case "INFO":
-		logger.Infof(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui)
-	case "WARNING":
-		logger.Warnf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui)
-	case "ERROR":
-		logger.Errorf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui)
-	default:
-		// 对于未知级别，使用 Warn 级别打印，并明确指出级别未知
-		logger.Warnf("ChirpStack 日志: [未知级别: %s] %s %s (DevEui: %s)",
-			event.Level, event.Code, event.Description, event.DeviceInfo.DevEui)
+	if err := c.handler.HandleStatus(reqCtx, event.DeviceInfo.DevEui, status); err != nil {
+		logger.Errorw("委托 'status' 事件失败", "error", err)
 	}
 }
 
-// handleJoinEvent 处理入网事件
-func (c *ChirpStackListener) handleJoinEvent(ctx context.Context, event *JoinEvent) {
-	logger := logs.TraceLogger(ctx, c.ctx, "handleJoinEvent")
-	logger.Infof("接收到 'join' 事件: %+v", event)
-	// 在这里添加您的业务逻辑
-}
+// adaptAckEvent 将 'ack' 事件适配并委托给 UpstreamHandler
+func (c *ChirpStackListener) adaptAckEvent(ctx context.Context, event *AckEvent) {
+	reqCtx, logger := logs.Trace(ctx, c.selfCtx, "adaptAckEvent")
 
-// handleTxAckEvent 处理网关发送确认事件
-func (c *ChirpStackListener) handleTxAckEvent(ctx context.Context, event *TxAckEvent) {
-	logger := logs.TraceLogger(ctx, c.ctx, "handleTxAckEvent")
-	logger.Infof("接收到 'txack' 事件: %+v", event)
-	// 在这里添加您的业务逻辑
-}
-
-// handleLocationEvent 处理位置事件
-func (c *ChirpStackListener) handleLocationEvent(ctx context.Context, event *LocationEvent) {
-	logger := logs.TraceLogger(ctx, c.ctx, "handleLocationEvent")
-	logger.Infof("接收到 'location' 事件: %+v", event)
-	// 在这里添加您的业务逻辑
-}
-
-// handleIntegrationEvent 处理集成事件
-func (c *ChirpStackListener) handleIntegrationEvent(ctx context.Context, event *IntegrationEvent) {
-	logger := logs.TraceLogger(ctx, c.ctx, "handleIntegrationEvent")
-	logger.Infof("接收到 'integration' 事件: %+v", event)
-	// 在这里添加您的业务逻辑
-}
-
-// recordSensorData 是一个通用方法，用于将传感器数据存入数据库。
-// areaControllerID: 区域主控设备的ID
-// sensorDeviceID: 实际产生传感器数据的普通设备的ID
-// sensorType: 传感器值的类型 (例如 models.SensorTypeTemperature)
-// data: 具体的传感器数据结构体实例 (例如 models.TemperatureData)
-func (c *ChirpStackListener) recordSensorData(ctx context.Context, areaControllerID uint32, sensorDeviceID uint32, eventTime time.Time, sensorType models.SensorType, data interface{}) {
-	reqCtx, logger := logs.Trace(ctx, c.ctx, "recordSensorData")
-	// 1. 将传入的结构体序列化为 JSON
-	jsonData, err := json.Marshal(data)
-	if err != nil {
-		logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err)
-		return
-	}
-
-	// 2. 构建 SensorData 模型
-	sensorData := &models.SensorData{
-		Time:             eventTime,
-		DeviceID:         sensorDeviceID,
-		AreaControllerID: areaControllerID,
-		SensorType:       sensorType,
-		Data:             datatypes.JSON(jsonData),
-	}
-
-	// 3. 调用仓库创建记录
-	if err := c.sensorDataRepo.Create(reqCtx, sensorData); err != nil {
-		logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err)
+	if err := c.handler.HandleAck(reqCtx, event.DeviceInfo.DevEui, event.DeduplicationID, event.Acknowledged, event.Time); err != nil {
+		logger.Errorw("委托 'ack' 事件失败", "error", err)
 	}
 }

internal/app/listener/lora_listener.go

@@ -0,0 +1,277 @@
+package listener
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math"
+	"time"
+
+	"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"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
+
+	"gorm.io/datatypes"
+)
+
+// loraListener 是一个统一的LoRa上行业务处理器，实现了 transport.UpstreamHandler 接口。
+// 它包含了处理业务所需的所有依赖，是项目中唯一处理LoRa上行业务的地方。
+type loraListener struct {
+	selfCtx               context.Context
+	areaControllerRepo    repository.AreaControllerRepository
+	pendingCollectionRepo repository.PendingCollectionRepository
+	deviceRepo            repository.DeviceRepository
+	sensorDataRepo        repository.SensorDataRepository
+	deviceCommandLogRepo  repository.DeviceCommandLogRepository
+}
+
+// NewLoRaListener 创建一个新的 loraListener 实例。
+// 注意：返回的是 transport.UpstreamHandler 接口，向上层隐藏具体实现。
+func NewLoRaListener(
+	ctx context.Context,
+	areaControllerRepo repository.AreaControllerRepository,
+	pendingCollectionRepo repository.PendingCollectionRepository,
+	deviceRepo repository.DeviceRepository,
+	sensorDataRepo repository.SensorDataRepository,
+	deviceCommandLogRepo repository.DeviceCommandLogRepository,
+) transport.UpstreamHandler {
+	return &loraListener{
+		selfCtx:               logs.AddCompName(ctx, "LoRaListener"),
+		areaControllerRepo:    areaControllerRepo,
+		pendingCollectionRepo: pendingCollectionRepo,
+		deviceRepo:            deviceRepo,
+		sensorDataRepo:        sensorDataRepo,
+		deviceCommandLogRepo:  deviceCommandLogRepo,
+	}
+}
+
+// HandleInstruction 处理来自设备的、已解析为Instruction的业务指令。
+func (l *loraListener) HandleInstruction(upstreamCtx context.Context, sourceAddr string, instruction *proto.Instruction) error {
+	ctx, logger := logs.Trace(upstreamCtx, l.selfCtx, "HandleInstruction")
+	logger.Infow("接收到设备指令", "来源地址", sourceAddr)
+
+	switch p := instruction.Payload.(type) {
+	case *proto.Instruction_CollectResult:
+		return l.handleCollectResult(ctx, sourceAddr, p.CollectResult)
+
+	case *proto.Instruction_OtaUpgradeStatus:
+		logger.Infow("收到OTA升级状态，暂未实现处理逻辑", "来源地址", sourceAddr, "状态", p.OtaUpgradeStatus)
+		// TODO: 在这里实现OTA升级状态的处理逻辑
+		return nil
+
+	case *proto.Instruction_LogUploadRequest:
+		logger.Infow("收到设备日志上传请求，暂未实现处理逻辑", "来源地址", sourceAddr, "日志条数", len(p.LogUploadRequest.Entries))
+		// TODO: 在这里实现设备日志的处理逻辑
+		return nil
+
+	default:
+		logger.Warnw("收到一个当前未处理的上行指令类型", "来源地址", sourceAddr, "类型", fmt.Sprintf("%T", p))
+		return nil
+	}
+}
+
+// HandleStatus 处理非业务指令的设备状态更新，例如信号强度、电量等。
+func (l *loraListener) HandleStatus(upstreamCtx context.Context, sourceAddr string, status map[string]interface{}) error {
+	ctx, logger := logs.Trace(upstreamCtx, l.selfCtx, "HandleStatus")
+	logger.Infow("接收到设备状态更新", "来源地址", sourceAddr, "状态", status)
+
+	areaController, err := l.areaControllerRepo.FindByNetworkID(ctx, sourceAddr)
+	if err != nil {
+		return fmt.Errorf("处理 'status' 事件失败：无法通过源地址 '%s' 找到区域主控设备: %w", sourceAddr, err)
+	}
+
+	eventTime := time.Now() // 状态事件通常是实时的，使用当前时间
+
+	// 尝试记录信号强度
+	if rssi, ok := status["rssi"].(float64); ok {
+		if snr, ok := status["snr"].(float64); ok {
+			signalMetrics := models.SignalMetrics{
+				RssiDbm: int(rssi),
+				SnrDb:   float32(snr),
+			}
+			if margin, ok := status["margin"].(float64); ok {
+				signalMetrics.MarginDb = int(margin)
+			}
+			l.recordSensorData(ctx, areaController.ID, areaController.ID, eventTime, models.SensorTypeSignalMetrics, signalMetrics)
+			logger.Infow("已记录区域主控的信号强度", "主控ID", areaController.ID, "指标", signalMetrics)
+		}
+	}
+
+	// 尝试记录电池电量
+	if batteryLevel, ok := status["batteryLevel"].(float64); ok {
+		batteryData := models.BatteryLevel{
+			BatteryLevelRatio: float32(batteryLevel),
+		}
+		if unavailable, ok := status["batteryLevelUnavailable"].(bool); ok {
+			batteryData.BatteryLevelUnavailable = unavailable
+		}
+		if externalPower, ok := status["externalPower"].(bool); ok {
+			batteryData.ExternalPower = externalPower
+		}
+		l.recordSensorData(ctx, areaController.ID, areaController.ID, eventTime, models.SensorTypeBatteryLevel, batteryData)
+		logger.Infow("已记录区域主控的电池状态", "主控ID", areaController.ID, "状态", batteryData)
+	}
+
+	return nil
+}
+
+// HandleAck 处理对下行指令的确认（ACK）事件。
+func (l *loraListener) HandleAck(upstreamCtx context.Context, sourceAddr string, deduplicationID string, acknowledged bool, eventTime time.Time) error {
+	ctx, logger := logs.Trace(upstreamCtx, l.selfCtx, "HandleAck")
+
+	err := l.deviceCommandLogRepo.UpdateAcknowledgedAt(ctx, deduplicationID, eventTime, acknowledged)
+	if err != nil {
+		logger.Errorw("更新下行任务记录的确认状态失败",
+			"MessageID", deduplicationID,
+			"DevEui", sourceAddr,
+			"Acknowledged", acknowledged,
+			"error", err,
+		)
+		return fmt.Errorf("更新下行任务记录失败: %w", err)
+	}
+
+	logger.Infow("成功更新下行任务记录确认状态",
+		"MessageID", deduplicationID,
+		"DevEui", sourceAddr,
+		"Acknowledged", acknowledged,
+		"AcknowledgedAt", eventTime.Format(time.RFC3339),
+	)
+	return nil
+}
+
+// handleCollectResult 是处理采集结果的核心业务逻辑
+func (l *loraListener) handleCollectResult(ctx context.Context, sourceAddr string, collectResp *proto.CollectResult) error {
+	if collectResp == nil {
+		return fmt.Errorf("传入的CollectResult为nil")
+	}
+
+	correlationID := collectResp.CorrelationId
+	logger := logs.GetLogger(ctx).With("correlationID", correlationID, "来源地址", sourceAddr)
+	logger.Infow("开始处理采集响应", "数据点数量", len(collectResp.Values))
+
+	// 1. 查找区域主控
+	areaController, err := l.areaControllerRepo.FindByNetworkID(ctx, sourceAddr)
+	if err != nil {
+		return fmt.Errorf("处理采集响应失败：无法通过源地址 '%s' 找到区域主控设备: %w", sourceAddr, err)
+	}
+	if err := areaController.SelfCheck(); err != nil {
+		return fmt.Errorf("处理采集响应失败：区域主控 %v(ID: %d) 未通过自检: %w", areaController.Name, areaController.ID, err)
+	}
+
+	// 2. 根据 CorrelationID 查找待处理请求
+	pendingReq, err := l.pendingCollectionRepo.FindByCorrelationID(ctx, correlationID)
+	if err != nil {
+		return fmt.Errorf("处理采集响应失败：无法找到待处理请求: %w", err)
+	}
+
+	// 3. 检查状态，防止重复处理
+	if pendingReq.Status != models.PendingStatusPending && pendingReq.Status != models.PendingStatusTimedOut {
+		logger.Warnw("收到一个已处理过的采集响应，将忽略。", "状态", string(pendingReq.Status))
+		return nil // 返回 nil，因为这不是一个错误，只是一个重复的请求
+	}
+
+	// 4. 匹配数据并存入数据库
+	deviceIDs := pendingReq.CommandMetadata
+	values := collectResp.Values
+	if len(deviceIDs) != len(values) {
+		err := fmt.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值", len(deviceIDs), len(values))
+		// 即使数量不匹配，也尝试更新状态为完成，以防止请求永远 pending
+		if updateErr := l.pendingCollectionRepo.UpdateStatusToFulfilled(ctx, correlationID, time.Now()); updateErr != nil {
+			logger.Errorw("更新待采集请求状态为 'fulfilled' 失败", "error", updateErr)
+		}
+		return err
+	}
+
+	eventTime := time.Now() // 对整个采集批次使用统一的时间戳
+
+	for i, deviceID := range deviceIDs {
+		rawSensorValue := values[i]
+		devLogger := logger.With("设备ID", deviceID)
+
+		if math.IsNaN(float64(rawSensorValue)) {
+			devLogger.Warnw("设备上报了一个无效的 NaN 值，已跳过当前值的记录。")
+			continue
+		}
+
+		dev, err := l.deviceRepo.FindByID(ctx, deviceID)
+		if err != nil {
+			devLogger.Errorw("处理采集数据失败：无法找到设备", "error", err)
+			continue
+		}
+		if err := dev.SelfCheck(); err != nil {
+			devLogger.Warnw("跳过设备，因其未通过自检或设备模板无效", "error", err)
+			continue
+		}
+
+		var valueDescriptors []*models.ValueDescriptor
+		if err := dev.DeviceTemplate.ParseValues(&valueDescriptors); err != nil {
+			devLogger.Warnw("跳过设备，因其设备模板的 Values 属性解析失败", "error", err)
+			continue
+		}
+		if len(valueDescriptors) == 0 {
+			devLogger.Warnw("跳过设备，因其设备模板缺少 ValueDescriptor 定义")
+			continue
+		}
+		valueDescriptor := valueDescriptors[0]
+
+		parsedValue := rawSensorValue*valueDescriptor.Multiplier + valueDescriptor.Offset
+
+		var dataToRecord interface{}
+		switch valueDescriptor.Type {
+		case models.SensorTypeTemperature:
+			dataToRecord = models.TemperatureData{TemperatureCelsius: parsedValue}
+		case models.SensorTypeHumidity:
+			dataToRecord = models.HumidityData{HumidityPercent: parsedValue}
+		case models.SensorTypeWeight:
+			dataToRecord = models.WeightData{WeightKilograms: parsedValue}
+		default:
+			devLogger.Warnw("未知的传感器类型，将使用通用格式记录", "传感器类型", string(valueDescriptor.Type))
+			dataToRecord = map[string]float32{"value": parsedValue}
+		}
+
+		l.recordSensorData(ctx, areaController.ID, dev.ID, eventTime, valueDescriptor.Type, dataToRecord)
+		devLogger.Infow("成功记录传感器数据",
+			"类型", string(valueDescriptor.Type),
+			"原始值", rawSensorValue,
+			"解析值", parsedValue,
+		)
+	}
+
+	// 5. 更新请求状态为“已完成”
+	if err := l.pendingCollectionRepo.UpdateStatusToFulfilled(ctx, correlationID, eventTime); err != nil {
+		logger.Errorw("更新待采集请求状态为 'fulfilled' 失败", "error", err)
+		return fmt.Errorf("更新待采集请求状态失败: %w", err)
+	}
+
+	logger.Infow("成功完成并关闭采集请求")
+	return nil
+}
+
+// recordSensorData 是一个通用方法，用于将传感器数据存入数据库。
+func (l *loraListener) recordSensorData(ctx context.Context, areaControllerID uint32, sensorDeviceID uint32, eventTime time.Time, sensorType models.SensorType, data interface{}) {
+	logger := logs.GetLogger(ctx).With("方法", "recordSensorData")
+
+	jsonData, err := json.Marshal(data)
+	if err != nil {
+		logger.Errorw("记录传感器数据失败：序列化数据为 JSON 时出错", "error", err)
+		return
+	}
+
+	sensorData := &models.SensorData{
+		Time:             eventTime,
+		DeviceID:         sensorDeviceID,
+		AreaControllerID: areaControllerID,
+		SensorType:       sensorType,
+		Data:             datatypes.JSON(jsonData),
+	}
+
+	if err := l.sensorDataRepo.Create(ctx, sensorData); err != nil {
+		logger.Errorw("记录传感器数据失败：存入数据库时出错",
+			"设备ID", sensorDeviceID,
+			"传感器类型", string(sensorType),
+			"error", err,
+		)
+	}
+}

internal/app/listener/transport.go

@@ -2,7 +2,7 @@ package listener
 
 import "net/http"
 
-// ListenHandler 是一个监听器, 用于监听设备上行事件
+// ListenHandler 是一个监听器, 用于监听设备上行事件, 通常用于适配http webhook。
 type ListenHandler interface {
 	Handler() http.HandlerFunc
 }