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/eclipse/paho.mqtt.golang"
)

type Service struct {
	cfg    config.Config
	keys   protocol.Keyring
	mqtt   mqtt.Client
	packet atomic.Uint32
	http   *http.Server
}
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
	}
	options := mqtt.NewClientOptions().AddBroker(cfg.MQTT.Broker).SetClientID(cfg.MQTT.ClientID).SetUsername(cfg.MQTT.Username).SetPassword(cfg.MQTT.Password).SetAutoReconnect(true).SetConnectRetry(true)
	if cfg.MQTT.TLS {
		tlsConfig, tlsErr := makeTLSConfig(cfg)
		if tlsErr != nil {
			return nil, tlsErr
		}
		options.SetTLSConfig(tlsConfig)
	}
	client := mqtt.NewClient(options)
	return &Service{cfg: cfg, keys: keys, mqtt: client}, 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 证书无效")
		}
	}
	result := &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots}
	if cfg.MQTT.CertificateFile != "" || cfg.MQTT.PrivateKeyFile != "" {
		certificate, loadErr := tls.LoadX509KeyPair(cfg.MQTT.CertificateFile, cfg.MQTT.PrivateKeyFile)
		if loadErr != nil {
			return nil, loadErr
		}
		result.Certificates = []tls.Certificate{certificate}
	}
	return result, nil
}

func (s *Service) Run(ctx context.Context) error {
	if token := s.mqtt.Connect(); !token.WaitTimeout(15 * time.Second) {
		return fmt.Errorf("MQTT 连接超时")
	} else if token.Error() != nil {
		return token.Error()
	}
	for _, topic := range []string{s.cfg.MQTT.UpTopic, s.cfg.MQTT.AckTopic} {
		if token := s.mqtt.Subscribe(topic, s.cfg.MQTT.QoS, s.onMessage); token.Wait() && token.Error() != nil {
			return token.Error()
		}
	}
	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.http.Shutdown(shutdown)
		s.mqtt.Disconnect(250)
	}()
	err := s.http.ListenAndServe()
	if err == http.ErrServerClosed {
		return nil
	}
	return err
}

func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
	if !s.mqtt.IsConnectionOpen() {
		http.Error(w, "mqtt disconnected", 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)
	token := s.mqtt.Publish(topic, s.cfg.MQTT.QoS, false, raw)
	if !token.WaitTimeout(10*time.Second) || token.Error() != 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, message mqtt.Message) {
	raw := append([]byte(nil), message.Payload()...)
	envelope := Envelope{Type: "device_message", Topic: message.Topic(), 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[:])
		envelope.DeviceID = deviceID
		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
	}
	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 := http.DefaultClient.Do(request)
	if err == nil {
		_ = response.Body.Close()
	}
}

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
}