diff --git a/AGENTS.md b/AGENTS.md index 5a9d319..0ae3ba1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,7 +30,7 @@ - `backend/api`：Gin HTTP API/BFF、鉴权、同步业务事务、数据库事实和审计。 - `backend/worker`：Outbox 投递、Redis Streams 消费、告警、派单、通知、对账和超时任务。当前主要是 Mock 边界，未确定持久化消息契约前不要伪造完整实现。 - `backend/iot-client`：管理系统/App 的设备上下行 HTTP 边界；不得绕过 API 的业务鉴权和安全状态校验。 -- `backend/iot-server`：外部 MQTT Broker 会话、厂商协议、遥测、设备命令和回执；本地配置不代表已完成真实设备联调。 +- `backend/iot-server`：内嵌 Mochi MQTT Broker、厂商协议、遥测、设备命令和回执；本地配置不代表已完成真实设备联调。 - `frontend/platform_admin`：平台级治理、运营、财务、安全、主数据和审计界面；后端资源契约是其资源和操作能力的依据。 - `docs`：产品和技术基线；涉及业务口径、实体、状态、权限或跨端流程的变更必须同步更新相应文档。 - `doc`：原始设计图和协议附件，仅作需求来源；不要批量改名、压缩或重写。 diff --git a/backend/README.md b/backend/README.md index af2d29a..a96bdd3 100644 --- a/backend/README.md +++ b/backend/README.md @@ -7,6 +7,6 @@ | `api` | Gin HTTP API / BFF、同步事务、JWT 鉴权与统一响应 | | `worker` | Outbox 投递、超时扫描与可重试异步任务 | | `iot-client` | 管理系统/App 的设备上下行 HTTP 边界、幂等受理与状态查询 | -| `iot-server` | 外部 MQTT Broker 会话、厂商二进制协议、遥测、命令与回执 | +| `iot-server` | 内嵌 Mochi MQTT Broker、厂商二进制协议、遥测、命令与回执 | 四个进程沿用统一的工程机制；API 是命令、Outbox 和上行报文的数据库事实主责，Worker 负责投递，IoT Client 负责系统接口，IoT Server 负责设备协议。 diff --git a/backend/go.work.sum b/backend/go.work.sum index 7a636ae..5d31711 100644 --- a/backend/go.work.sum +++ b/backend/go.work.sum @@ -13,8 +13,6 @@ github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= -github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= -github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= github.com/elastic/elastic-transport-go/v8 v8.9.0 h1:KeT/2P54F0xS0S8Y3Pf+tFDg4HmBgReQMB+BMz8dDAs= github.com/elastic/elastic-transport-go/v8 v8.9.0/go.mod h1:ssMTvNS2hwf7CaiGsRRsx4gQHFZ/jS/DkLcISxekWzc= github.com/elastic/go-elasticsearch/v9 v9.4.1 h1:pEF8xlnL8D2WdVp4HRHhGNUwC5QHdgk0DZKrJY0WCNs= @@ -35,8 +33,6 @@ github.com/godbus/dbus/v5 v5.0.4 h1:9349emZab16e7zQvpmsbtjc18ykshndd8y2PG3sgJbA= github.com/golang/glog v1.2.5 h1:DrW6hGnjIhtvhOIiAKT6Psh/Kd/ldepEa81DKeiRJ5I= github.com/golang/glog v1.2.5/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w= github.com/google/gofuzz v1.0.0 h1:A8PeW59pxE9IoFRqBp37U+mSNaQoZ46F1f0f863XSXw= -github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= -github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1 h1:qnpSQwGEnkcRpTqNOIR6bJbR0gAorgP9CSALpRcKoAA= github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1/go.mod h1:lXGCsh6c22WGtjr+qGHj1otzZpV/1kwTMAqkwZsnWRU= github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 h1:pRhl55Yx1eC7BZ1N+BBWwnKaMyD8uC+34TLdndZMAKk= @@ -71,6 +67,8 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= github.com/rogpeppe/fastuuid v1.2.0 h1:Ppwyp6VYCF1nvBTXL3trRso7mXMlRrw9ooo375wvi2s= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= +github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY= +github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo= diff --git a/backend/iot-server/README.md b/backend/iot-server/README.md index b6cfde0..89ed623 100644 --- a/backend/iot-server/README.md +++ b/backend/iot-server/README.md @@ -1,5 +1,5 @@ # IoT Server -设备侧 MQTT 适配进程。连接外部 EMQX/Mosquitto Broker，订阅 `devices/+/up` 与 `devices/+/ack`，按《气体探测器通讯协议》V1.8 校验、解密和解析原始二进制帧；下行命令由受保护的内部 HTTP 接口接收后发布到 `devices/{deviceId}/down`。 +设备侧 MQTT 接入进程。使用 `github.com/mochi-mqtt/server/v2` 内嵌实现 MQTT Broker，监听 `devices/+/up` 与 `devices/+/ack`，按《气体探测器通讯协议》V1.8 校验、解密和解析原始二进制帧；下行命令由受保护的内部 HTTP 接口接收后发布到 `devices/{deviceId}/down`，不依赖 Eclipse Paho 或外部 EMQX/Mosquitto。 -生产环境必须启用 MQTT TLS、每设备凭证并通过环境变量或密钥注入覆盖 YAML 开发占位密钥。 +YAML 的 `MQTT.Devices` 为每台设备配置独立 Client ID、设备号和凭证，Broker ACL 只允许设备发布自己的 `up/ack` 并订阅自己的 `down`。生产环境必须启用 MQTT TLS，并通过密钥注入覆盖 YAML 开发占位密码；单设备开发配置可用 `HEQI_IOT_MQTT_PASSWORD` 覆盖密码。 diff --git a/backend/iot-server/etc/platform_iot_server_dev.yaml b/backend/iot-server/etc/platform_iot_server_dev.yaml index 71bfdc8..555b885 100644 --- a/backend/iot-server/etc/platform_iot_server_dev.yaml +++ b/backend/iot-server/etc/platform_iot_server_dev.yaml @@ -1,9 +1,12 @@ Service: platform-iot-server MQTT: - Broker: tcp://127.0.0.1:1883 - ClientID: heqi-iot-server-dev - Username: heqi-dev - Password: change-me + Address: 127.0.0.1:1883 + ListenerID: mqtt-tcp + Devices: + - ClientID: detector-0000000000000001 + DeviceID: "0000000000000001" + Username: detector-0000000000000001 + Password: change-me UpTopic: devices/+/up DownTopic: devices/{deviceId}/down AckTopic: devices/+/ack diff --git a/backend/iot-server/go.mod b/backend/iot-server/go.mod index 2641db8..94769ca 100644 --- a/backend/iot-server/go.mod +++ b/backend/iot-server/go.mod @@ -3,12 +3,11 @@ module git.apinb.com/heqiapp/platforms/backend/iot-server go 1.26.1 require ( - github.com/eclipse/paho.mqtt.golang v1.5.1 + github.com/mochi-mqtt/server/v2 v2.7.9 gopkg.in/yaml.v3 v3.0.1 ) require ( github.com/gorilla/websocket v1.5.3 // indirect - golang.org/x/net v0.52.0 // indirect - golang.org/x/sync v0.20.0 // indirect + github.com/rs/xid v1.4.0 // indirect ) diff --git a/backend/iot-server/go.sum b/backend/iot-server/go.sum index 74aed36..b82a23a 100644 --- a/backend/iot-server/go.sum +++ b/backend/iot-server/go.sum @@ -1,11 +1,9 @@ -github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE= -github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= -golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +github.com/mochi-mqtt/server/v2 v2.7.9 h1:y0g4vrSLAag7T07l2oCzOa/+nKVLoazKEWAArwqBNYI= +github.com/mochi-mqtt/server/v2 v2.7.9/go.mod h1:lZD3j35AVNqJL5cezlnSkuG05c0FCHSsfAKSPBOSbqc= +github.com/rs/xid v1.4.0 h1:qd7wPTDkN6KQx2VmMBLrpHkiyQwgFXRnkOLacUiaSNY= +github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/backend/iot-server/internal/config/config.go b/backend/iot-server/internal/config/config.go index 60aeb9f..fa9bb56 100644 --- a/backend/iot-server/internal/config/config.go +++ b/backend/iot-server/internal/config/config.go @@ -11,18 +11,23 @@ import ( ) type MQTT struct { - Broker string `yaml:"Broker"` - ClientID string `yaml:"ClientID"` - Username string `yaml:"Username"` - Password string `yaml:"Password"` - UpTopic string `yaml:"UpTopic"` - DownTopic string `yaml:"DownTopic"` - AckTopic string `yaml:"AckTopic"` - QoS byte `yaml:"QoS"` - TLS bool `yaml:"TLS"` - CAFile string `yaml:"CAFile"` - CertificateFile string `yaml:"CertificateFile"` - PrivateKeyFile string `yaml:"PrivateKeyFile"` + Address string `yaml:"Address"` + ListenerID string `yaml:"ListenerID"` + Devices []MQTTDevice `yaml:"Devices"` + UpTopic string `yaml:"UpTopic"` + DownTopic string `yaml:"DownTopic"` + AckTopic string `yaml:"AckTopic"` + QoS byte `yaml:"QoS"` + TLS bool `yaml:"TLS"` + CAFile string `yaml:"CAFile"` + CertificateFile string `yaml:"CertificateFile"` + PrivateKeyFile string `yaml:"PrivateKeyFile"` +} +type MQTTDevice struct { + ClientID string `yaml:"ClientID"` + DeviceID string `yaml:"DeviceID"` + Username string `yaml:"Username"` + Password string `yaml:"Password"` } type HTTP struct { Address string `yaml:"Address"` @@ -50,13 +55,26 @@ func Load(path string) (Config, error) { if err = yaml.Unmarshal(data, &cfg); err != nil { return cfg, err } - override(&cfg.MQTT.Password, "HEQI_IOT_MQTT_PASSWORD") + if len(cfg.MQTT.Devices) == 1 { + override(&cfg.MQTT.Devices[0].Password, "HEQI_IOT_MQTT_PASSWORD") + } override(&cfg.HTTP.InternalToken, "HEQI_IOT_INTERNAL_TOKEN") override(&cfg.Protocol.Key1, "HEQI_IOT_KEY_1") override(&cfg.Protocol.Key2, "HEQI_IOT_KEY_2") override(&cfg.Protocol.Key3, "HEQI_IOT_KEY_3") - if cfg.MQTT.Broker == "" || cfg.HTTP.Address == "" || cfg.HTTP.InternalToken == "" { - return cfg, fmt.Errorf("MQTT.Broker、HTTP.Address 和 HTTP.InternalToken 必填") + if cfg.MQTT.Address == "" || cfg.HTTP.Address == "" || cfg.HTTP.InternalToken == "" { + return cfg, fmt.Errorf("MQTT.Address、HTTP.Address 和 HTTP.InternalToken 必填") + } + if cfg.MQTT.ListenerID == "" { + cfg.MQTT.ListenerID = "mqtt-tcp" + } + if len(cfg.MQTT.Devices) == 0 { + return cfg, fmt.Errorf("MQTT.Devices 必填，Broker 禁止匿名连接") + } + for index, device := range cfg.MQTT.Devices { + if device.ClientID == "" || device.DeviceID == "" || device.Username == "" || device.Password == "" { + return cfg, fmt.Errorf("MQTT.Devices[%d] 的 ClientID、DeviceID、Username 和 Password 必填", index) + } } return cfg, nil } diff --git a/backend/iot-server/internal/config/config_test.go b/backend/iot-server/internal/config/config_test.go new file mode 100644 index 0000000..b9e37ea --- /dev/null +++ b/backend/iot-server/internal/config/config_test.go @@ -0,0 +1,34 @@ +package config + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadRequiresDeviceCredentials(t *testing.T) { + path := filepath.Join(t.TempDir(), "iot.yaml") + data := []byte("MQTT:\n Address: 127.0.0.1:1883\nHTTP:\n Address: 127.0.0.1:12428\n InternalToken: test\n") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + if _, err := Load(path); err == nil { + t.Fatal("未配置设备凭证时应拒绝启动") + } +} + +func TestLoadDefaultsListenerAndOverridesSingleDevicePassword(t *testing.T) { + t.Setenv("HEQI_IOT_MQTT_PASSWORD", "from-secret-store") + path := filepath.Join(t.TempDir(), "iot.yaml") + data := []byte("MQTT:\n Address: 127.0.0.1:1883\n Devices:\n - ClientID: detector-1\n DeviceID: '0000000000000001'\n Username: detector-1\n Password: placeholder\nHTTP:\n Address: 127.0.0.1:12428\n InternalToken: test\n") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if cfg.MQTT.ListenerID != "mqtt-tcp" || cfg.MQTT.Devices[0].Password != "from-secret-store" { + t.Fatalf("unexpected MQTT config: %+v", cfg.MQTT) + } +} diff --git a/backend/iot-server/internal/service/service.go b/backend/iot-server/internal/service/service.go index eec365c..56af9b4 100644 --- a/backend/iot-server/internal/service/service.go +++ b/backend/iot-server/internal/service/service.go @@ -1,4 +1,4 @@ -// Package service 连接外部 MQTT Broker，并在内部 HTTP 边界接收待下发命令。 +// Package service 内嵌 MQTT Broker，并在内部 HTTP 边界接收待下发命令。 package service import ( @@ -19,15 +19,19 @@ import ( "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" + 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 - mqtt mqtt.Client + broker *mqtt.Server packet atomic.Uint32 http *http.Server + ready atomic.Bool } type Command struct { Identity string `json:"identity"` @@ -63,16 +67,37 @@ func New(cfg config.Config) (*Service, error) { 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) + 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 } - options.SetTLSConfig(tlsConfig) + listenerConfig.TLSConfig = tlsConfig } - client := mqtt.NewClient(options) - return &Service{cfg: cfg, keys: keys, mqtt: client}, nil + if err = broker.AddListener(listeners.NewTCP(listenerConfig)); err != nil { + return nil, fmt.Errorf("配置 MQTT 监听器: %w", err) + } + return &Service{cfg: cfg, keys: keys, broker: broker}, nil } func makeTLSConfig(cfg config.Config) (*tls.Config, error) { @@ -89,28 +114,32 @@ func makeTLSConfig(cfg config.Config) (*tls.Config, error) { 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} + 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 token := s.mqtt.Connect(); !token.WaitTimeout(15 * time.Second) { - return fmt.Errorf("MQTT 连接超时") - } else if token.Error() != nil { - return token.Error() + if err := s.broker.Subscribe(s.cfg.MQTT.UpTopic, 1, s.onMessage); err != nil { + return fmt.Errorf("订阅设备上行 Topic: %w", err) } - 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() - } + 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) @@ -119,8 +148,9 @@ func (s *Service) Run(ctx context.Context) error { <-ctx.Done() shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() + s.ready.Store(false) _ = s.http.Shutdown(shutdown) - s.mqtt.Disconnect(250) + _ = s.broker.Close() }() err := s.http.ListenAndServe() if err == http.ErrServerClosed { @@ -130,8 +160,8 @@ func (s *Service) Run(ctx context.Context) error { } func (s *Service) health(w http.ResponseWriter, _ *http.Request) { - if !s.mqtt.IsConnectionOpen() { - http.Error(w, "mqtt disconnected", http.StatusServiceUnavailable) + if !s.ready.Load() { + http.Error(w, "mqtt broker unavailable", http.StatusServiceUnavailable) return } w.WriteHeader(http.StatusNoContent) @@ -181,8 +211,7 @@ func (s *Service) command(w http.ResponseWriter, r *http.Request) { 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 { + if err := s.broker.Publish(topic, raw, false, s.cfg.MQTT.QoS); err != nil { http.Error(w, "mqtt publish failed", http.StatusBadGateway) return } @@ -191,9 +220,9 @@ func (s *Service) command(w http.ResponseWriter, r *http.Request) { _ = 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)} +func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message packets.Packet) { + raw := append([]byte(nil), message.Payload...) + envelope := Envelope{Type: "device_message", Topic: message.TopicName, 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 diff --git a/checking/README.md b/checking/README.md index 0759151..188853a 100644 --- a/checking/README.md +++ b/checking/README.md @@ -24,6 +24,6 @@ ## 计划边界 -- `backend/worker` 已投递 IoT Outbox；`backend/iot-client` 与 `backend/iot-server` 已落地系统接口和 MQTT 协议边界。真实 Broker、设备证书和厂商硬件仍需在联调环境验收。 +- `backend/worker` 已投递 IoT Outbox；`backend/iot-client` 与内嵌 Mochi MQTT Broker 的 `backend/iot-server` 已落地系统接口和 MQTT 协议边界。真实设备证书和厂商硬件仍需在联调环境验收。 - `frontend/site` 是官网，按展示、响应式、可访问性、链接、构建和托管 Worker 测试；不把它当成业务事实写入端。 - 本计划不把尚未实现或文档中“待确认”的政策当作通过标准。此类项进入阻塞/待决清单，由产品、安全、财务或法务确认后再固化用例。 diff --git a/docs/10-技术实现规划.md b/docs/10-技术实现规划.md index 2182289..29fe738 100644 --- a/docs/10-技术实现规划.md +++ b/docs/10-技术实现规划.md @@ -54,7 +54,7 @@ flowchart LR | `api` | 用户端和五个管理系统的 HTTP API、鉴权、同步业务事务 | 无状态部署；所有写操作支持幂等键、事务和审计 | | `worker` | 事件消费、告警、派单、推送、对账、超时扫描、轨迹异常识别 | Redis Streams 消费组；重试、死信、幂等消费和可观测的积压告警 | | `iot-client` | 管理系统/App 的设备上下行 HTTP 边界、幂等受理与状态查询 | 不绕过 API 的 JWT、角色、对象归属与安全状态校验 | -| `iot-server` | 外部 MQTT Broker 会话、厂商协议适配、遥测校验、命令下发与回执 | 保持设备会话一致性；协议版本、设备身份、LRC8 与 AES 校验 | +| `iot-server` | 内嵌 Mochi MQTT Broker、厂商协议适配、遥测校验、命令下发与回执 | 保持设备会话一致性；每设备认证与 Topic ACL；协议版本、设备身份、LRC8 与 AES 校验 | 关键业务采用“数据库事务 + Outbox 事件表 + Worker 投递”的模式：先在 PostgreSQL 提交业务事实与待投递事件，再异步写入 Redis Streams。这样 Redis 故障或 Worker 重启不会丢失订单、告警、支付或设备命令的业务事实。 diff --git a/docs/11-数据接口与安全.md b/docs/11-数据接口与安全.md index a0c8239..c1be48f 100644 --- a/docs/11-数据接口与安全.md +++ b/docs/11-数据接口与安全.md @@ -44,7 +44,7 @@ ## 3. IoT 协议与可靠性 - 设备厂商 V1.8 二进制帧保持 `0x5E` 起始、`0x5B` 结束、大端序、数据包 AES-128 与 LRC8 规则不变，并作为 MQTT payload 传输；Topic 使用 `devices/{deviceId}/{up|down|ack}`，QoS 1，下行命令禁止 retained。 -- `iot-server` 只处理 MQTT 会话和设备协议，`iot-client` 只提供系统侧上下行接口；命令、Outbox、原始上行和回执事实由 API 持久化，Worker 负责可重试投递。 +- `iot-server` 使用 Mochi MQTT v2 内嵌 Broker，只处理 MQTT 会话、每设备认证/Topic ACL 和设备协议；`iot-client` 只提供系统侧上下行接口；命令、Outbox、原始上行和回执事实由 API 持久化，Worker 负责可重试投递。 - 协议封面版本与变更记录冲突时以最新 V1.8 变更记录和绿色标注为兼容实现依据；重复子标识等歧义必须保留原始报文并按设备型号配置解析，不得静默猜测。 - 设备采用 MQTT over TLS，设备身份使用每设备证书或短期轮换令牌；禁止共享默认密钥。