lc_garage/internal/service/parking/mqtt_gate_controller.go

package parking

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

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gofrs/uuid/v5"
	"go.uber.org/zap"
	"gorm.io/gorm"
	"wails-app/internal/dao"
	"wails-app/internal/global"
	incidentService "wails-app/internal/modules/incident/service"
)

const (
	mqttQoS        = 1
	mqttCmdTimeout = 5 * time.Second
)

// gateTopicRoute 设备在 MQTT 主题树中的路由坐标。
type gateTopicRoute struct {
	DeviceCode string
	LotID      uint
	BoothID    uint
	ChannelID  uint
}

func (r gateTopicRoute) cmdTopic() string {
	return fmt.Sprintf("parking/lot/%d/booth/%d/channel/%d/gate/cmd", r.LotID, r.BoothID, r.ChannelID)
}

// MQTTGateController 通过 MQTT 下发道闸指令并等待设备 ack，实现 GateController 接口。
// 用于「独立网络道闸控制器」；挂在 UHF 读写器继电器上的道闸仍走 DeviceManager。
type MQTTGateController struct {
	client mqtt.Client

	mu              sync.RWMutex
	routes          map[string]gateTopicRoute // device_code -> route
	connectionState map[string]bool           // device_code -> MQTT 连接状态
	gateState       map[string]string         // device_code -> open/closed/moving/fault
	pending         map[string]pendingGateCommand
}

type pendingGateCommand struct {
	deviceCode string
	ackCh      chan gateAck
}

type gateCmd struct {
	Schema     string `json:"schema"`
	CmdID      string `json:"cmd_id"`
	DeviceCode string `json:"device_code"`
	Action     string `json:"action"`
	ValidTime  byte   `json:"valid_time"`
}

type gateAck struct {
	Schema     string `json:"schema"`
	CmdID      string `json:"cmd_id"`
	DeviceCode string `json:"device_code"`
	Result     string `json:"result"` // success / failed
	Error      string `json:"error"`
}

func NewMQTTGateController(client mqtt.Client) *MQTTGateController {
	return &MQTTGateController{
		client:          client,
		routes:          make(map[string]gateTopicRoute),
		connectionState: make(map[string]bool),
		gateState:       make(map[string]string),
		pending:         make(map[string]pendingGateCommand),
	}
}

// SetClient 在 MQTT 客户端完成选项配置并创建后注入控制器。
// 初始化阶段尚未开始订阅或下发指令，因此这里不需要额外同步。
func (c *MQTTGateController) SetClient(client mqtt.Client) {
	c.mu.Lock()
	c.client = client
	c.mu.Unlock()
}

// LoadRoutes 从 DB 反查 device_code -> lot/booth/channel，用于构造主题。
func (c *MQTTGateController) LoadRoutes() error {
	if global.GVA_DB == nil {
		return errors.New("数据库未初始化")
	}
	type row struct {
		DeviceCode   string
		ChannelID    uint
		BoothID      uint
		ParkingLotID uint
	}
	var rows []row
	err := global.GVA_DB.Model(&dao.UHFReader{}).
		Select("uhf_reader.device_code, uhf_reader.channel_id, channel.booth_id, channel.parking_lot_id").
		Joins("LEFT JOIN channel ON channel.id = uhf_reader.channel_id").
		Where("uhf_reader.channel_id > 0 AND channel.booth_id > 0 AND channel.parking_lot_id > 0 AND uhf_reader.connect_type = ?", dao.ConnectTypeMQTT).
		Scan(&rows).Error
	if err != nil {
		return err
	}
	c.mu.Lock()
	// 配置变更后重新加载，不能保留已删除设备的旧路由。
	c.routes = make(map[string]gateTopicRoute)
	defer c.mu.Unlock()
	for _, r := range rows {
		if r.DeviceCode == "" || r.ChannelID == 0 || r.BoothID == 0 || r.ParkingLotID == 0 {
			continue
		}
		c.routes[r.DeviceCode] = gateTopicRoute{
			DeviceCode: r.DeviceCode, ChannelID: r.ChannelID,
			BoothID: r.BoothID, LotID: r.ParkingLotID,
		}
	}
	return nil
}

// HasRoute 判断设备是否配置了 MQTT 路由（供路由控制器决定走哪个后端）。
func (c *MQTTGateController) HasRoute(deviceCode string) bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	_, ok := c.routes[deviceCode]
	return ok
}

func (c *MQTTGateController) routeOf(deviceCode string) (gateTopicRoute, bool) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	r, ok := c.routes[deviceCode]
	return r, ok
}

// OpenGate 实现 GateController 接口。
func (c *MQTTGateController) OpenGate(deviceCode string, validTime byte) error {
	return c.run(deviceCode, "open", validTime)
}

// CloseGate 实现 GateController 接口。
func (c *MQTTGateController) CloseGate(deviceCode string, validTime byte) error {
	return c.run(deviceCode, "close", validTime)
}

func (c *MQTTGateController) run(deviceCode, action string, validTime byte) error {
	route, ok := c.routeOf(deviceCode)
	if !ok {
		return fmt.Errorf("设备未配置 MQTT 路由: %s", deviceCode)
	}
	cmdID := uuid.Must(uuid.NewV4()).String()
	ackCh := make(chan gateAck, 1)
	c.mu.Lock()
	c.pending[cmdID] = pendingGateCommand{deviceCode: deviceCode, ackCh: ackCh}
	c.mu.Unlock()
	defer func() {
		c.mu.Lock()
		delete(c.pending, cmdID)
		c.mu.Unlock()
	}()

	payload, _ := json.Marshal(gateCmd{
		Schema: "gate.cmd.v1", CmdID: cmdID, DeviceCode: deviceCode,
		Action: action, ValidTime: validTime,
	})
	token := c.client.Publish(route.cmdTopic(), mqttQoS, false, payload)
	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("MQTT 指令下发失败: %w", token.Error())
	}

	select {
	case ack := <-ackCh:
		if ack.Result != "success" {
			return fmt.Errorf("道闸执行失败: %s", ack.Error)
		}
		return nil
	case <-time.After(mqttCmdTimeout):
		return errors.New("道闸指令超时未收到 ack")
	}
}

// IsGateConnected 实现 GateController 接口（以最近 state/lwt 为准）。
func (c *MQTTGateController) IsGateConnected(deviceCode string) bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.connectionState[deviceCode]
}

// Subscribe 订阅 state / ack / lwt 主题。
func (c *MQTTGateController) Subscribe() error {
	for _, topic := range []string{
		"parking/lot/+/booth/+/channel/+/gate/state",
		"parking/lot/+/booth/+/channel/+/gate/cmd/ack",
		"parking/lot/+/booth/+/channel/+/gate/lwt",
	} {
		token := c.client.Subscribe(topic, mqttQoS, c.onMessage)
		if token.Wait() && token.Error() != nil {
			return token.Error()
		}
	}
	return nil
}

// MarkConfiguredDevicesOffline 在服务启动或 MQTT 总线重连期间清除旧进程留下的在线状态。
// 后续收到设备的 retained state 或实时状态上报后，会立即恢复为在线。
// 仅处理启用的 MQTT 设备，不能影响 TCP、串口设备自身的连接状态。
func (c *MQTTGateController) MarkConfiguredDevicesOffline() error {
	if global.GVA_DB == nil {
		return errors.New("数据库未初始化")
	}
	return global.GVA_DB.Model(&dao.UHFReader{}).
		Where("connect_type = ? AND is_active = ?", dao.ConnectTypeMQTT, true).
		Where("status <> ?", "offline").
		Update("status", "offline").Error
}

// onMessage 分派 state / ack / lwt。
func (c *MQTTGateController) onMessage(_ mqtt.Client, msg mqtt.Message) {
	payload := msg.Payload()
	var probe struct {
		Schema     string `json:"schema"`
		CmdID      string `json:"cmd_id"`
		DeviceCode string `json:"device_code"`
		State      string `json:"state"`
	}
	if err := json.Unmarshal(payload, &probe); err != nil {
		return
	}

	if global.GVA_LOG != nil {
		global.GVA_LOG.Info("收到道闸 MQTT 消息",
			zap.String("topic", msg.Topic()),
			zap.String("schema", probe.Schema),
			zap.String("device_code", probe.DeviceCode),
			zap.String("state", probe.State))
	}

	switch probe.Schema {
	case "gate.ack.v1":
		var ack gateAck
		if json.Unmarshal(payload, &ack) == nil && ack.CmdID != "" && ack.DeviceCode != "" {
			c.mu.RLock()
			pending, ok := c.pending[ack.CmdID]
			c.mu.RUnlock()
			if ok && pending.deviceCode == ack.DeviceCode && ack.Schema == "gate.ack.v1" {
				select {
				case pending.ackCh <- ack:
				default:
				}
			}
		}
	case "gate.state.v1":
		if probe.DeviceCode == "" || probe.State == "" {
			return
		}
		c.mu.Lock()
		c.connectionState[probe.DeviceCode] = true
		c.gateState[probe.DeviceCode] = probe.State
		c.mu.Unlock()
		// 回写设备在线状态并关闭历史离线异常。
		// 设备编码必须与设备管理中的 device_code 一致，否则只能更新内存状态。
		c.persistDeviceStatus(probe.DeviceCode, "online")
		if global.GVA_DB != nil {
			incidentSvc.ResolveDeviceOffline(probe.DeviceCode)
		}
	case "gate.lwt.v1", "gate.lwt":
		c.mu.Lock()
		c.connectionState[probe.DeviceCode] = false
		c.mu.Unlock()
		// 回写设备离线状态并埋点离线异常（复用既有去重）
		c.persistDeviceStatus(probe.DeviceCode, "offline")
		if global.GVA_DB != nil {
			incidentSvc.RecordIncident(incidentService.RecordIncidentRequest{
				Category:    incidentService.CategoryDeviceOffline,
				Source:      incidentService.SourceDevice,
				DeviceCode:  probe.DeviceCode,
				Description: "道闸 MQTT 遗嘱触发，设备离线",
			})
		}
	}
}

// persistDeviceStatus 将 MQTT 设备状态写回设备表，并显式暴露编码不匹配问题。
// MQTT 订阅本身是按主题成功的，即使设备尚未在设备管理中配置，消息仍会进入这里；
// 此时 GORM 不会报错，只会返回 RowsAffected=0，不能继续静默处理。
func (c *MQTTGateController) persistDeviceStatus(deviceCode, status string) {
	if global.GVA_DB == nil || deviceCode == "" {
		return
	}
	updates := map[string]interface{}{"status": status}
	if status == "online" {
		updates["provision_status"] = gorm.Expr("CASE WHEN connect_type = ? AND provision_status IN (?, ?, ?, ?) THEN ? ELSE provision_status END", dao.ConnectTypeMQTT, "provisioned", "online", "offline", "failed", "online")
	} else if status == "offline" {
		updates["provision_status"] = gorm.Expr("CASE WHEN connect_type = ? AND provision_status IN (?, ?) THEN ? ELSE provision_status END", dao.ConnectTypeMQTT, "provisioned", "online", "offline")
	}
	if status == "online" {
		updates["last_online_time"] = time.Now()
	}
	result := global.GVA_DB.Model(&dao.UHFReader{}).
		Where("device_code = ?", deviceCode).
		Updates(updates)
	if result.Error != nil {
		if global.GVA_LOG != nil {
			global.GVA_LOG.Error("回写 MQTT 设备状态失败",
				zap.String("device_code", deviceCode), zap.String("status", status), zap.Error(result.Error))
		}
		return
	}
	if result.RowsAffected == 0 && global.GVA_LOG != nil {
		global.GVA_LOG.Warn("收到 MQTT 设备状态，但设备编码未配置或不匹配",
			zap.String("device_code", deviceCode), zap.String("status", status))
	}
}