lc_garage/internal/service/devicebus/device_bus.go

package devicebus

import (
	"encoding/json"
	"fmt"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"go.uber.org/zap"
	"wails-app/internal/dao"
	"wails-app/internal/global"
	common "wails-app/internal/model/common"
	parking "wails-app/internal/service/parking"
	"wails-app/internal/service/uhf"
)

// DeviceBus 订阅相机识别事件，并将事件转发到统一通行入口。
// 道闸状态、ACK 和 LWT 由 MQTTGateController 统一处理，避免重叠订阅覆盖回调。
type DeviceBus struct {
	passage *parking.PassageService
}

func NewDeviceBus() *DeviceBus {
	return &DeviceBus{passage: &parking.PassageService{}}
}

// CameraEvent 是相机识别事件的 MQTT 数据结构。
type CameraEvent struct {
	Schema      string  `json:"schema"`
	DeviceCode  string  `json:"device_code"`
	PlateNumber string  `json:"plate_number"`
	RFIDTag     string  `json:"rfid_tag"`
	Direction   string  `json:"direction"`
	TS          int64   `json:"ts"`
	ImagePath   string  `json:"image_path"`
	ImageID     string  `json:"image_id"`
	EventID     string  `json:"event_id"`
	Confidence  float64 `json:"confidence"`
}

// Subscribe 只订阅相机事件，设备状态类消息由道闸控制器订阅。
func (b *DeviceBus) Subscribe(client mqtt.Client) error {
	topic := "parking/lot/+/booth/+/channel/+/camera/event"
	token := client.Subscribe(topic, 1, b.onMessage)
	if token.Wait() && token.Error() != nil {
		return token.Error()
	}
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("MQTT 设备主题订阅成功", zap.String("topic", topic))
	}
	return nil
}

func (b *DeviceBus) onMessage(_ mqtt.Client, msg mqtt.Message) {
	var probe struct {
		Schema string `json:"schema"`
	}
	if err := json.Unmarshal(msg.Payload(), &probe); err != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Warn("MQTT 设备消息 JSON 无效", zap.String("topic", msg.Topic()), zap.Error(err))
		}
		return
	}
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("收到设备 MQTT 消息", zap.String("topic", msg.Topic()), zap.String("schema", probe.Schema))
	}
	if probe.Schema == "camera.event.v1" {
		// 识别处理包含数据库事务和等待道闸 ACK，异步执行避免阻塞 MQTT 回调。
		// 回调返回后 MQTT 客户端可以复用消息缓冲区，因此协程必须使用副本。
		payload := append([]byte(nil), msg.Payload()...)
		if global.GVA_LOG != nil {
			global.GVA_LOG.Info("收到车辆识别 MQTT 事件", zap.String("topic", msg.Topic()), zap.Int("payload_bytes", len(payload)))
		}
		go b.onCameraEvent(payload)
	}
}

func (b *DeviceBus) onCameraEvent(payload []byte) {
	startedAt := time.Now()
	var ev CameraEvent
	if err := json.Unmarshal(payload, &ev); err != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Warn("车辆识别 MQTT 事件解析失败", zap.Int("payload_bytes", len(payload)), zap.Error(err))
		}
		return
	}
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("车辆识别事件字段已解析", zap.String("device_code", ev.DeviceCode), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.String("plate_number", ev.PlateNumber), zap.String("direction", ev.Direction), zap.Int64("ts", ev.TS))
	}
	imagePath, err := resolveImagePath(ev.DeviceCode, ev.ImageID, ev.ImagePath)
	if err != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Warn("车辆识别事件引用的图片不可用", zap.String("device_code", ev.DeviceCode), zap.String("image_id", ev.ImageID), zap.String("event_id", ev.EventID), zap.Error(err))
		}
		return
	}
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("车辆识别事件图片引用校验成功", zap.String("device_code", ev.DeviceCode), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.Bool("legacy_image_path", ev.ImageID == ""))
	}
	req := common.PassageRequest{
		DeviceCode: ev.DeviceCode, PlateNumber: ev.PlateNumber, RFIDTag: ev.RFIDTag,
		Direction: ev.Direction, Image: imagePath, TriggerSource: "camera",
	}
	result, err := b.passage.HandlePassage(req)
	status := "completed"
	message := "识别通过"
	fee := 0.0
	stayTime := int64(0)
	if err != nil {
		status = "failed"
		message = err.Error()
	} else if result != nil {
		fee = result.Fee
		stayTime = result.StayTime
		if result.Message != "" {
			message = result.Message
		}
	}
	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("处理车辆识别事件",
			zap.String("device_code", ev.DeviceCode), zap.String("plate_number", ev.PlateNumber),
			zap.String("direction", ev.Direction), zap.String("event_id", ev.EventID), zap.String("image_id", ev.ImageID), zap.String("status", status), zap.String("message", message), zap.Duration("duration", time.Since(startedAt)))
	}
	uhf.PushChannelEvent(uhf.ChannelEvent{
		DeviceCode: ev.DeviceCode, RFIDTag: ev.RFIDTag, PlateNumber: ev.PlateNumber,
		Direction: ev.Direction, Timestamp: ev.TS, Status: status, Message: message,
		Fee: fee, StayTime: stayTime,
	})
}

// resolveImagePath accepts the new uploaded-image reference. image_path stays
// supported for deployed legacy cameras until their firmware is upgraded.
func resolveImagePath(deviceCode, imageID, legacyPath string) (string, error) {
	if imageID == "" {
		return legacyPath, nil
	}
	if global.GVA_DB == nil {
		return "", fmt.Errorf("数据库未初始化")
	}
	var image dao.DeviceImage
	if err := global.GVA_DB.Where("device_code = ? AND image_id = ?", deviceCode, imageID).First(&image).Error; err != nil {
		return "", fmt.Errorf("图片未上传或不属于该设备")
	}
	return image.URL, nil
}