From bd4f5b83e12ae5d0fa2ba4cd8b8ad6cd7b7d0b5b Mon Sep 17 00:00:00 2001 From: huang <1724659546@qq.com> Date: Wed, 5 Nov 2025 18:54:24 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9webhook=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/app/webhook/chirp_stack.go | 191 ++++++++++--------- internal/app/webhook/placeholder_listener.go | 11 +- 2 files changed, 108 insertions(+), 94 deletions(-) diff --git a/internal/app/webhook/chirp_stack.go b/internal/app/webhook/chirp_stack.go index 7be960e..1af2d83 100644 --- a/internal/app/webhook/chirp_stack.go +++ b/internal/app/webhook/chirp_stack.go @@ -1,6 +1,7 @@ package webhook import ( + "context" "encoding/base64" "encoding/json" "io" @@ -30,7 +31,7 @@ const ( // ChirpStackListener 是一个监听器, 用于监听ChirpStack反馈的设备上行事件 type ChirpStackListener struct { - logger *logs.Logger + ctx context.Context sensorDataRepo repository.SensorDataRepository deviceRepo repository.DeviceRepository areaControllerRepo repository.AreaControllerRepository @@ -40,15 +41,15 @@ type ChirpStackListener struct { // NewChirpStackListener 创建一个新的 ChirpStackListener 实例 func NewChirpStackListener( - logger *logs.Logger, + ctx context.Context, sensorDataRepo repository.SensorDataRepository, deviceRepo repository.DeviceRepository, areaControllerRepo repository.AreaControllerRepository, deviceCommandLogRepo repository.DeviceCommandLogRepository, pendingCollectionRepo repository.PendingCollectionRepository, -) ListenHandler { // 返回接口类型 +) ListenHandler { return &ChirpStackListener{ - logger: logger, + ctx: ctx, sensorDataRepo: sensorDataRepo, deviceRepo: deviceRepo, areaControllerRepo: areaControllerRepo, @@ -60,11 +61,13 @@ func NewChirpStackListener( // 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") + defer r.Body.Close() b, err := io.ReadAll(r.Body) if err != nil { - c.logger.Errorf("读取请求体失败: %v", err) + logger.Errorf("读取请求体失败: %v", err) http.Error(w, "failed to read body", http.StatusBadRequest) return } @@ -74,100 +77,102 @@ func (c *ChirpStackListener) Handler() http.HandlerFunc { w.WriteHeader(http.StatusOK) // 将异步处理逻辑委托给 handler 方法 - go c.handler(b, event) + go c.handler(ctx, b, event) } } // handler 用于处理 ChirpStack 发送的事件 -func (c *ChirpStackListener) handler(data []byte, eventType string) { +func (c *ChirpStackListener) handler(ctx context.Context, data []byte, eventType string) { + reqCtx, logger := logs.Trace(ctx, c.ctx, "ChirpStackListener.handler") switch eventType { case eventTypeUp: var msg UpEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'up' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'up' 事件失败: %v, data: %s", err, string(data)) return } - c.handleUpEvent(&msg) + c.handleUpEvent(reqCtx, &msg) case eventTypeJoin: var msg JoinEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'join' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'join' 事件失败: %v, data: %s", err, string(data)) return } - c.handleJoinEvent(&msg) + c.handleJoinEvent(reqCtx, &msg) case eventTypeAck: var msg AckEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'ack' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'ack' 事件失败: %v, data: %s", err, string(data)) return } - c.handleAckEvent(&msg) + c.handleAckEvent(reqCtx, &msg) case eventTypeTxAck: var msg TxAckEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'txack' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'txack' 事件失败: %v, data: %s", err, string(data)) return } - c.handleTxAckEvent(&msg) + c.handleTxAckEvent(reqCtx, &msg) case eventTypeStatus: var msg StatusEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'status' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'status' 事件失败: %v, data: %s", err, string(data)) return } - c.handleStatusEvent(&msg) + c.handleStatusEvent(reqCtx, &msg) case eventTypeLog: var msg LogEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'log' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'log' 事件失败: %v, data: %s", err, string(data)) return } - c.handleLogEvent(&msg) + c.handleLogEvent(reqCtx, &msg) case eventTypeLocation: var msg LocationEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'location' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'location' 事件失败: %v, data: %s", err, string(data)) return } - c.handleLocationEvent(&msg) + c.handleLocationEvent(reqCtx, &msg) case eventTypeIntegration: var msg IntegrationEvent if err := json.Unmarshal(data, &msg); err != nil { - c.logger.Errorf("解析 'integration' 事件失败: %v, data: %s", err, string(data)) + logger.Errorf("解析 'integration' 事件失败: %v, data: %s", err, string(data)) return } - c.handleIntegrationEvent(&msg) + c.handleIntegrationEvent(reqCtx, &msg) default: - c.logger.Errorf("未知的ChirpStack事件: %s, data: %s", eventType, string(data)) + logger.Errorf("未知的ChirpStack事件: %s, data: %s", eventType, string(data)) } } // --- 业务处理函数 --- // handleUpEvent 处理上行数据事件 -func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { - c.logger.Infof("开始处理 'up' 事件, DevEui: %s", event.DeviceInfo.DevEui) +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) // 1. 查找区域主控设备 - regionalController, err := c.areaControllerRepo.FindByNetworkID(event.DeviceInfo.DevEui) + regionalController, err := c.areaControllerRepo.FindByNetworkID(reqCtx, event.DeviceInfo.DevEui) if err != nil { - c.logger.Errorf("处理 'up' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err) + logger.Errorf("处理 'up' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err) return } // 依赖 SelfCheck 确保区域主控有效 if err := regionalController.SelfCheck(); err != nil { - c.logger.Errorf("处理 'up' 事件失败：区域主控 %v(ID: %d) 未通过自检: %v", regionalController.Name, regionalController.ID, err) + logger.Errorf("处理 'up' 事件失败：区域主控 %v(ID: %d) 未通过自检: %v", regionalController.Name, regionalController.ID, err) return } - c.logger.Infof("找到区域主控: %s (ID: %d)", regionalController.Name, regionalController.ID) + logger.Infof("找到区域主控: %s (ID: %d)", regionalController.Name, regionalController.ID) // 2. 记录区域主控的信号强度 (如果存在) if len(event.RxInfo) > 0 { @@ -182,29 +187,29 @@ func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { } // 记录信号强度 - c.recordSensorData(regionalController.ID, regionalController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics) - c.logger.Infof("已记录区域主控 (ID: %d) 的信号强度: RSSI=%d, SNR=%.2f", regionalController.ID, rx.Rssi, rx.Snr) + c.recordSensorData(reqCtx, regionalController.ID, regionalController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics) + logger.Infof("已记录区域主控 (ID: %d) 的信号强度: RSSI=%d, SNR=%.2f", regionalController.ID, rx.Rssi, rx.Snr) } else { - c.logger.Warnf("处理 'up' 事件时未找到 RxInfo，无法记录信号数据。DevEui: %s", event.DeviceInfo.DevEui) + logger.Warnf("处理 'up' 事件时未找到 RxInfo，无法记录信号数据。DevEui: %s", event.DeviceInfo.DevEui) } // 3. 处理上报的传感器数据 if event.Data == "" { - c.logger.Warnf("处理 'up' 事件时 Data 字段为空，无需记录上行数据。DevEui: %s", event.DeviceInfo.DevEui) + logger.Warnf("处理 'up' 事件时 Data 字段为空，无需记录上行数据。DevEui: %s", event.DeviceInfo.DevEui) return } // 3.1 Base64 解码 decodedData, err := base64.StdEncoding.DecodeString(event.Data) if err != nil { - c.logger.Errorf("Base64 解码 'up' 事件的 Data 失败: %v, Data: %s", err, event.Data) + logger.Errorf("Base64 解码 'up' 事件的 Data 失败: %v, Data: %s", err, event.Data) return } // 3.2 解析外层 "信封" var instruction proto.Instruction if err := gproto.Unmarshal(decodedData, &instruction); err != nil { - c.logger.Errorf("解析上行 Instruction Protobuf 失败: %v, Decoded Data: %x", err, decodedData) + logger.Errorf("解析上行 Instruction Protobuf 失败: %v, Decoded Data: %x", err, decodedData) return } @@ -215,29 +220,29 @@ func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { collectResp = p.CollectResult default: // 如果上行的数据不是采集结果，记录日志并忽略 - c.logger.Infof("收到一个非采集响应的上行指令 (Type: %T)，无需处理。", p) + logger.Infof("收到一个非采集响应的上行指令 (Type: %T)，无需处理。", p) return } // 检查 collectResp 是否为 nil，虽然在 type switch 成功的情况下不太可能 if collectResp == nil { - c.logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil") + logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil") return } correlationID := collectResp.CorrelationId - c.logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values)) + logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values)) // 4. 根据 CorrelationID 查找待处理请求 - pendingReq, err := c.pendingCollectionRepo.FindByCorrelationID(correlationID) + pendingReq, err := c.pendingCollectionRepo.FindByCorrelationID(reqCtx, correlationID) if err != nil { - c.logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err) + logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err) return } // 检查状态，防止重复处理 if pendingReq.Status != models.PendingStatusPending && pendingReq.Status != models.PendingStatusTimedOut { - c.logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status) + logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status) return } @@ -245,11 +250,11 @@ func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { deviceIDs := pendingReq.CommandMetadata values := collectResp.Values if len(deviceIDs) != len(values) { - c.logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID) + logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID) // 即使数量不匹配，也更新状态为完成，以防止请求永远 pending - err = c.pendingCollectionRepo.UpdateStatusToFulfilled(correlationID, event.Time) + err = c.pendingCollectionRepo.UpdateStatusToFulfilled(reqCtx, correlationID, event.Time) if err != nil { - c.logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err) + logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err) } return } @@ -259,35 +264,35 @@ func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { // 检查设备上报的值是否为 NaN (Not a Number)，如果是则跳过 if math.IsNaN(float64(rawSensorValue)) { - c.logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID) + logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID) continue } // 5.1 获取设备及其模板 - dev, err := c.deviceRepo.FindByID(deviceID) + dev, err := c.deviceRepo.FindByID(reqCtx, deviceID) if err != nil { - c.logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err) + logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err) continue } // 依赖 SelfCheck 确保设备和模板有效 if err := dev.SelfCheck(); err != nil { - c.logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err) continue } if err := dev.DeviceTemplate.SelfCheck(); err != nil { - c.logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err) continue } // 5.2 从设备模板中解析 ValueDescriptor var valueDescriptors []*models.ValueDescriptor if err := dev.DeviceTemplate.ParseValues(&valueDescriptors); err != nil { - c.logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err) continue } // 根据 DeviceTemplate.SelfCheck，这里应该只有一个 ValueDescriptor if len(valueDescriptors) == 0 { - c.logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID) + logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID) continue } valueDescriptor := valueDescriptors[0] @@ -306,31 +311,32 @@ func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { dataToRecord = models.WeightData{WeightKilograms: parsedValue} default: // TODO 未知传感器的数据需要记录吗 - c.logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type) + logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type) dataToRecord = map[string]float64{"value": parsedValue} } // 5.5 记录传感器数据 - c.recordSensorData(regionalController.ID, dev.ID, event.Time, valueDescriptor.Type, dataToRecord) - c.logger.Infof("成功记录传感器数据: 设备ID=%d, 类型=%s, 原始值=%f, 解析值=%.2f", dev.ID, valueDescriptor.Type, rawSensorValue, parsedValue) + c.recordSensorData(reqCtx, regionalController.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(correlationID, event.Time); err != nil { - c.logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err) + if err := c.pendingCollectionRepo.UpdateStatusToFulfilled(reqCtx, correlationID, event.Time); err != nil { + logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err) } else { - c.logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID) + logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID) } } // handleStatusEvent 处理设备状态事件 -func (c *ChirpStackListener) handleStatusEvent(event *StatusEvent) { - c.logger.Infof("处接收到理 'status' 事件: %+v", event) +func (c *ChirpStackListener) handleStatusEvent(ctx context.Context, event *StatusEvent) { + reqCtx, logger := logs.Trace(ctx, c.ctx, "handleStatusEvent") + logger.Infof("处接收到理 'status' 事件: %+v", event) // 查找区域主控设备 - regionalController, err := c.areaControllerRepo.FindByNetworkID(event.DeviceInfo.DevEui) + regionalController, err := c.areaControllerRepo.FindByNetworkID(reqCtx, event.DeviceInfo.DevEui) if err != nil { - c.logger.Errorf("处理 'status' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err) + logger.Errorf("处理 'status' 事件失败：无法通过 DevEui '%s' 找到区域主控设备: %v", event.DeviceInfo.DevEui, err) return } @@ -338,8 +344,8 @@ func (c *ChirpStackListener) handleStatusEvent(event *StatusEvent) { signalMetrics := models.SignalMetrics{ MarginDb: event.Margin, } - c.recordSensorData(regionalController.ID, regionalController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics) - c.logger.Infof("已记录区域主控 (ID: %d) 的信号状态: %+v", regionalController.ID, signalMetrics) + c.recordSensorData(reqCtx, regionalController.ID, regionalController.ID, event.Time, models.SensorTypeSignalMetrics, signalMetrics) + logger.Infof("已记录区域主控 (ID: %d) 的信号状态: %+v", regionalController.ID, signalMetrics) // 记录电量 batteryLevel := models.BatteryLevel{ @@ -347,68 +353,74 @@ func (c *ChirpStackListener) handleStatusEvent(event *StatusEvent) { BatteryLevelUnavailable: event.BatteryLevelUnavailable, ExternalPower: event.ExternalPower, } - c.recordSensorData(regionalController.ID, regionalController.ID, event.Time, models.SensorTypeBatteryLevel, batteryLevel) - c.logger.Infof("已记录区域主控 (ID: %d) 的电池状态: %+v", regionalController.ID, batteryLevel) + c.recordSensorData(reqCtx, regionalController.ID, regionalController.ID, event.Time, models.SensorTypeBatteryLevel, batteryLevel) + logger.Infof("已记录区域主控 (ID: %d) 的电池状态: %+v", regionalController.ID, batteryLevel) } // handleAckEvent 处理下行确认事件 -func (c *ChirpStackListener) handleAckEvent(event *AckEvent) { - c.logger.Infof("接收到 'ack' 事件: %+v", event) +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(event.DeduplicationID, event.Time, event.Acknowledged) + err := c.deviceCommandLogRepo.UpdateAcknowledgedAt(reqCtx, event.DeduplicationID, event.Time, event.Acknowledged) if err != nil { - c.logger.Errorf("更新下行任务记录的确认时间及接收成功状态失败 (MessageID: %s, DevEui: %s, Acknowledged: %t): %v", + logger.Errorf("更新下行任务记录的确认时间及接收成功状态失败 (MessageID: %s, DevEui: %s, Acknowledged: %t): %v", event.DeduplicationID, event.DeviceInfo.DevEui, event.Acknowledged, err) return } - c.logger.Infof("成功更新下行任务记录确认时间及接收成功状态 (MessageID: %s, DevEui: %s, Acknowledged: %t, AcknowledgedAt: %s)", + 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(event *LogEvent) { +func (c *ChirpStackListener) handleLogEvent(ctx context.Context, event *LogEvent) { + logger := logs.TraceLogger(ctx, c.ctx, "handleLogEvent") // 首先，打印完整的事件结构体，用于详细排查 - c.logger.Infof("接收到 'log' 事件的完整内容: %+v", event) + logger.Infof("接收到 'log' 事件的完整内容: %+v", event) // 接着，根据 ChirpStack 日志的级别，使用我们自己的 logger 对应级别来打印核心信息 logMessage := "ChirpStack 日志: [%s] %s (DevEui: %s)" switch event.Level { case "INFO": - c.logger.Infof(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) + logger.Infof(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) case "WARNING": - c.logger.Warnf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) + logger.Warnf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) case "ERROR": - c.logger.Errorf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) + logger.Errorf(logMessage, event.Code, event.Description, event.DeviceInfo.DevEui) default: // 对于未知级别，使用 Warn 级别打印，并明确指出级别未知 - c.logger.Warnf("ChirpStack 日志: [未知级别: %s] %s %s (DevEui: %s)", + logger.Warnf("ChirpStack 日志: [未知级别: %s] %s %s (DevEui: %s)", event.Level, event.Code, event.Description, event.DeviceInfo.DevEui) } } // handleJoinEvent 处理入网事件 -func (c *ChirpStackListener) handleJoinEvent(event *JoinEvent) { - c.logger.Infof("接收到 'join' 事件: %+v", event) +func (c *ChirpStackListener) handleJoinEvent(ctx context.Context, event *JoinEvent) { + logger := logs.TraceLogger(ctx, c.ctx, "handleJoinEvent") + logger.Infof("接收到 'join' 事件: %+v", event) // 在这里添加您的业务逻辑 } // handleTxAckEvent 处理网关发送确认事件 -func (c *ChirpStackListener) handleTxAckEvent(event *TxAckEvent) { - c.logger.Infof("接收到 'txack' 事件: %+v", event) +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(event *LocationEvent) { - c.logger.Infof("接收到 'location' 事件: %+v", event) +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(event *IntegrationEvent) { - c.logger.Infof("接收到 'integration' 事件: %+v", event) +func (c *ChirpStackListener) handleIntegrationEvent(ctx context.Context, event *IntegrationEvent) { + logger := logs.TraceLogger(ctx, c.ctx, "handleIntegrationEvent") + logger.Infof("接收到 'integration' 事件: %+v", event) // 在这里添加您的业务逻辑 } @@ -417,11 +429,12 @@ func (c *ChirpStackListener) handleIntegrationEvent(event *IntegrationEvent) { // sensorDeviceID: 实际产生传感器数据的普通设备的ID // sensorType: 传感器值的类型 (例如 models.SensorTypeTemperature) // data: 具体的传感器数据结构体实例 (例如 models.TemperatureData) -func (c *ChirpStackListener) recordSensorData(regionalControllerID uint, sensorDeviceID uint, eventTime time.Time, sensorType models.SensorType, data interface{}) { +func (c *ChirpStackListener) recordSensorData(ctx context.Context, regionalControllerID uint, sensorDeviceID uint, 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 { - c.logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err) + logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err) return } @@ -435,7 +448,7 @@ func (c *ChirpStackListener) recordSensorData(regionalControllerID uint, sensorD } // 3. 调用仓库创建记录 - if err := c.sensorDataRepo.Create(sensorData); err != nil { - c.logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err) + if err := c.sensorDataRepo.Create(reqCtx, sensorData); err != nil { + logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err) } } diff --git a/internal/app/webhook/placeholder_listener.go b/internal/app/webhook/placeholder_listener.go index ffcf216..2d3979d 100644 --- a/internal/app/webhook/placeholder_listener.go +++ b/internal/app/webhook/placeholder_listener.go @@ -1,6 +1,7 @@ package webhook import ( + "context" "net/http" "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" @@ -8,15 +9,14 @@ import ( // PlaceholderListener 是一个占位符, 用于在非 LoRaWAN 配置下满足 ListenHandler 接口 type PlaceholderListener struct { - logger *logs.Logger + ctx context.Context } // NewPlaceholderListener 创建一个新的 PlaceholderListener 实例 // 它只打印一条日志, 表明 ChirpStack webhook 未被激活 -func NewPlaceholderListener(logger *logs.Logger) ListenHandler { - logger.Info("当前配置非 LoRaWAN, ChirpStack webhook 监听器未激活。") +func NewPlaceholderListener(ctx context.Context) ListenHandler { return &PlaceholderListener{ - logger: logger, + ctx: ctx, } } @@ -24,7 +24,8 @@ func NewPlaceholderListener(logger *logs.Logger) ListenHandler { // 理论上, 在占位符生效的模式下, 这个 Handler 不应该被调用 func (p *PlaceholderListener) Handler() http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { - p.logger.Warn("PlaceholderListener 的 Handler 被调用, 这通常是意料之外的。") + logger := logs.TraceLogger(r.Context(), p.ctx, "PlaceholderListener") + logger.Warn("PlaceholderListener 的 Handler 被调用, 这通常是意料之外的。") w.WriteHeader(http.StatusNotImplemented) } }