platforms/backend/iot-server/internal/service/service.go

// Package service 内嵌 MQTT Broker，并在内部 HTTP 边界接收待下发命令。
package service

import (
	"bytes"
	"context"
	"crypto/subtle"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"sync/atomic"
	"time"

	"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/config"
	"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/protocol"
	mqtt "github.com/mochi-mqtt/server/v2"
	"github.com/mochi-mqtt/server/v2/hooks/auth"
	"github.com/mochi-mqtt/server/v2/listeners"
	"github.com/mochi-mqtt/server/v2/packets"
)

type Service struct {
	cfg    config.Config
	keys   protocol.Keyring
	broker *mqtt.Server
	packet atomic.Uint32
	http   *http.Server
	ready  atomic.Bool
	client *http.Client
}
type Command struct {
	Identity       string    `json:"identity"`
	IdempotencyKey string    `json:"idempotency_key"`
	DeviceID       string    `json:"device_id"`
	Action         string    `json:"action"`
	ExpiresAt      time.Time `json:"expires_at"`
	KeyID          byte      `json:"key_id"`
	DeviceKind     byte      `json:"device_kind"`
	DeviceType     byte      `json:"device_type"`
	DeviceModel    [3]byte   `json:"device_model"`
	Controller     byte      `json:"controller"`
	Loop           byte      `json:"loop"`
	Component      byte      `json:"component"`
}
type Envelope struct {
	Type, Topic, DeviceID, ReceivedAt string
	PayloadHex                        string
	Frame                             *DecodedFrame `json:"frame,omitempty"`
}
type DecodedFrame struct {
	KeyID, Version, Control, MainID byte
	PacketNumber                    uint16
	Sequence                        byte
	Final                           bool
	DeviceTime                      string
	PayloadHex                      string
	Realtime                        *protocol.RealtimeData `json:"realtime,omitempty"`
}

func New(cfg config.Config) (*Service, error) {
	keys, err := cfg.Keys()
	if err != nil {
		return nil, err
	}
	broker := mqtt.New(&mqtt.Options{InlineClient: true})
	ledger := &auth.Ledger{}
	for _, device := range cfg.MQTT.Devices {
		ledger.Auth = append(ledger.Auth, auth.AuthRule{
			Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
			Password: auth.RString(device.Password), Allow: true,
		})
		ledger.ACL = append(ledger.ACL, auth.ACLRule{
			Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
			Filters: auth.Filters{
				auth.RString(strings.ReplaceAll(cfg.MQTT.UpTopic, "+", device.DeviceID)):            auth.WriteOnly,
				auth.RString(strings.ReplaceAll(cfg.MQTT.AckTopic, "+", device.DeviceID)):           auth.WriteOnly,
				auth.RString(strings.ReplaceAll(cfg.MQTT.DownTopic, "{deviceId}", device.DeviceID)): auth.ReadOnly,
			},
		})
	}
	if err = broker.AddHook(new(auth.Hook), &auth.Options{Ledger: ledger}); err != nil {
		return nil, fmt.Errorf("配置 MQTT 认证: %w", err)
	}
	listenerConfig := listeners.Config{ID: cfg.MQTT.ListenerID, Address: cfg.MQTT.Address}
	if cfg.MQTT.TLS {
		tlsConfig, tlsErr := makeTLSConfig(cfg)
		if tlsErr != nil {
			return nil, tlsErr
		}
		listenerConfig.TLSConfig = tlsConfig
	}
	if err = broker.AddListener(listeners.NewTCP(listenerConfig)); err != nil {
		return nil, fmt.Errorf("配置 MQTT 监听器: %w", err)
	}
	return &Service{cfg: cfg, keys: keys, broker: broker, client: &http.Client{Timeout: 12 * time.Second}}, nil
}

func makeTLSConfig(cfg config.Config) (*tls.Config, error) {
	roots, err := x509.SystemCertPool()
	if err != nil {
		roots = x509.NewCertPool()
	}
	if cfg.MQTT.CAFile != "" {
		data, readErr := os.ReadFile(cfg.MQTT.CAFile)
		if readErr != nil {
			return nil, readErr
		}
		if !roots.AppendCertsFromPEM(data) {
			return nil, fmt.Errorf("MQTT CA 证书无效")
		}
	}
	if cfg.MQTT.CertificateFile == "" || cfg.MQTT.PrivateKeyFile == "" {
		return nil, fmt.Errorf("启用 MQTT TLS 时 CertificateFile 和 PrivateKeyFile 必填")
	}
	certificate, err := tls.LoadX509KeyPair(cfg.MQTT.CertificateFile, cfg.MQTT.PrivateKeyFile)
	if err != nil {
		return nil, err
	}
	result := &tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{certificate}}
	if cfg.MQTT.CAFile != "" {
		result.ClientCAs = roots
		result.ClientAuth = tls.RequireAndVerifyClientCert
	}
	return result, nil
}

func (s *Service) Run(ctx context.Context) error {
	if err := s.broker.Subscribe(s.cfg.MQTT.UpTopic, 1, s.onMessage); err != nil {
		return fmt.Errorf("订阅设备上行 Topic: %w", err)
	}
	if err := s.broker.Subscribe(s.cfg.MQTT.AckTopic, 2, s.onMessage); err != nil {
		return fmt.Errorf("订阅设备回执 Topic: %w", err)
	}
	if err := s.broker.Serve(); err != nil {
		return fmt.Errorf("启动 MQTT Broker: %w", err)
	}
	s.ready.Store(true)
	mux := http.NewServeMux()
	mux.HandleFunc("/health", s.health)
	mux.HandleFunc("/internal/v1/commands", s.command)
	s.http = &http.Server{Addr: s.cfg.HTTP.Address, Handler: mux, ReadHeaderTimeout: 5 * time.Second}
	go func() {
		<-ctx.Done()
		shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		s.ready.Store(false)
		_ = s.http.Shutdown(shutdown)
		_ = s.broker.Close()
	}()
	err := s.http.ListenAndServe()
	if err == http.ErrServerClosed {
		return nil
	}
	return err
}

func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
	if !s.ready.Load() {
		http.Error(w, "mqtt broker unavailable", http.StatusServiceUnavailable)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

func (s *Service) command(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	if !secureEqual(r.Header.Get("X-Heqi-Iot-Token"), s.cfg.HTTP.InternalToken) {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}
	var cmd Command
	if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&cmd); err != nil {
		http.Error(w, "invalid json", http.StatusBadRequest)
		return
	}
	if cmd.Identity == "" || cmd.IdempotencyKey == "" || cmd.DeviceID == "" || time.Now().After(cmd.ExpiresAt) {
		http.Error(w, "invalid or expired command", http.StatusUnprocessableEntity)
		return
	}
	deviceID, err := decodeDeviceID(cmd.DeviceID)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	var payload []byte
	switch cmd.Action {
	case "open_valve":
		payload, err = protocol.ValveCommand(cmd.Controller, cmd.Loop, cmd.Component, true)
	case "close_valve":
		payload, err = protocol.ValveCommand(cmd.Controller, cmd.Loop, cmd.Component, false)
	default:
		err = fmt.Errorf("unsupported action")
	}
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	n := s.packet.Add(1)
	frame := protocol.Frame{KeyID: cmd.KeyID, Version: 1, Control: 0x10, DeviceKind: cmd.DeviceKind, DeviceType: cmd.DeviceType, DeviceModel: cmd.DeviceModel, DeviceID: deviceID, PacketNumber: uint16(n%65535 + 1), Sequence: 1, Final: true, DeviceTime: time.Now(), Payload: payload}
	raw, err := protocol.Encode(frame, s.keys)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnprocessableEntity)
		return
	}
	topic := strings.ReplaceAll(s.cfg.MQTT.DownTopic, "{deviceId}", cmd.DeviceID)
	if err := s.broker.Publish(topic, raw, false, s.cfg.MQTT.QoS); err != nil {
		http.Error(w, "mqtt publish failed", http.StatusBadGateway)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(map[string]any{"command_identity": cmd.Identity, "packet_number": frame.PacketNumber, "status": "dispatched"})
}

func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message packets.Packet) {
	raw := append([]byte(nil), message.Payload...)
	expectedDeviceID, topicOK := deviceIDFromTopic(s.cfg, message.TopicName)
	envelope := Envelope{Type: "device_message", Topic: message.TopicName, DeviceID: expectedDeviceID, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
	if frame, err := protocol.Decode(raw, s.keys); err == nil {
		deviceID := hex.EncodeToString(frame.DeviceID[:])
		if !topicOK || deviceID != expectedDeviceID {
			s.forwardEnvelope(envelope)
			return
		}
		main := byte(0)
		if len(frame.Payload) > 0 {
			main = frame.Payload[0]
		}
		decoded := &DecodedFrame{KeyID: frame.KeyID, Version: frame.Version, Control: frame.Control, MainID: main, PacketNumber: frame.PacketNumber, Sequence: frame.Sequence, Final: frame.Final, DeviceTime: frame.DeviceTime.Format(time.RFC3339), PayloadHex: hex.EncodeToString(frame.Payload)}
		if main == protocol.MainRealtime {
			if realtime, parseErr := protocol.DecodeRealtime(frame.Payload); parseErr == nil {
				decoded.Realtime = &realtime
			}
		}
		envelope.Frame = decoded
	}
	s.forwardEnvelope(envelope)
}

func (s *Service) forwardEnvelope(envelope Envelope) {
	data, _ := json.Marshal(envelope)
	request, err := http.NewRequest(http.MethodPost, s.cfg.HTTP.CallbackURL, bytes.NewReader(data))
	if err != nil {
		return
	}
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("X-Heqi-Iot-Token", s.cfg.HTTP.InternalToken)
	response, err := s.client.Do(request)
	if err == nil {
		_ = response.Body.Close()
	}
}

func deviceIDFromTopic(cfg config.Config, topic string) (string, bool) {
	for _, pattern := range []string{cfg.MQTT.UpTopic, cfg.MQTT.AckTopic} {
		parts := strings.Split(pattern, "/")
		values := strings.Split(topic, "/")
		if len(parts) != len(values) {
			continue
		}
		deviceID := ""
		matched := true
		for index := range parts {
			if parts[index] == "+" {
				deviceID = values[index]
				continue
			}
			if parts[index] != values[index] {
				matched = false
				break
			}
		}
		if matched && len(deviceID) == 16 {
			return strings.ToLower(deviceID), true
		}
	}
	return "", false
}

func decodeDeviceID(value string) ([8]byte, error) {
	var result [8]byte
	decoded, err := hex.DecodeString(value)
	if err != nil || len(decoded) != 8 {
		return result, fmt.Errorf("device_id 必须是 16 位 BCD/十六进制字符串")
	}
	copy(result[:], decoded)
	return result, nil
}
func secureEqual(left, right string) bool {
	return len(left) == len(right) && subtle.ConstantTimeCompare([]byte(left), []byte(right)) == 1
}