From c1406137186d857c81180eb3643d385d05e3bb03 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 22 Jun 2026 17:25:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(M06-A):=20=E5=BB=BA=E7=AB=8BMQTT=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E4=B8=8EBroker=E5=9F=BA=E7=A1=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/.env.example | 4 + backend/package-lock.json | 448 +++++++++++++++++++++++++- backend/package.json | 3 +- backend/src/app.ts | 5 +- backend/src/config.ts | 24 +- backend/src/mqtt/mqtt-service.ts | 171 ++++++++++ backend/src/routes/health.ts | 36 ++- backend/src/server.ts | 5 + backend/tests/mqtt-service.test.mjs | 113 +++++++ backend/tests/payment.test.mjs | 2 + backend/tests/profit-sharing.test.mjs | 2 + deploy/emqx/acl.conf.template | 28 +- scripts/dev/wsl/mqtt-smoke.sh | 11 +- scripts/setup/emqx.sh | 57 ++++ setup.sh | 2 +- 15 files changed, 875 insertions(+), 36 deletions(-) create mode 100644 backend/src/mqtt/mqtt-service.ts create mode 100644 backend/tests/mqtt-service.test.mjs diff --git a/backend/.env.example b/backend/.env.example index 991364c..662c7e1 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,5 +17,9 @@ QIPAI_WECHAT_PAY_CREDENTIALS={} QIPAI_PROFIT_SHARE_MOCK_ENABLED=false QIPAI_THIRD_PARTY_CREDENTIALS={} QIPAI_MQTT_URL=mqtt://101.42.38.246:1883 +QIPAI_MQTT_CLIENT_ID=qipai-backend QIPAI_MQTT_USERNAME= QIPAI_MQTT_PASSWORD= +QIPAI_MQTT_RECONNECT_MS=3000 +QIPAI_MQTT_CONNECT_TIMEOUT_MS=10000 +QIPAI_MQTT_MAX_MESSAGE_BYTES=65536 diff --git a/backend/package-lock.json b/backend/package-lock.json index 5d2f335..bc7bd20 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -11,6 +11,7 @@ "@fastify/cors": "^11.2.0", "@fastify/rate-limit": "^11.0.0", "fastify": "^5.8.5", + "mqtt": "^5.15.1", "mysql2": "^3.11.3", "pino": "^9.4.0", "sharp": "^0.34.0", @@ -26,6 +27,14 @@ "npm": ">=10.0.0" } }, + "node_modules/@babel/runtime": { + "version": "7.29.7", + "resolved": "https://registry.npmjs.org/@babel/runtime/-/runtime-7.29.7.tgz", + "integrity": "sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw==", + "engines": { + "node": ">=6.9.0" + } + }, "node_modules/@emnapi/runtime": { "version": "1.11.1", "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", @@ -973,6 +982,33 @@ "undici-types": "~6.21.0" } }, + "node_modules/@types/readable-stream": { + "version": "4.0.23", + "resolved": "https://registry.npmjs.org/@types/readable-stream/-/readable-stream-4.0.23.tgz", + "integrity": "sha512-wwXrtQvbMHxCbBgjHaMGEmImFTQxxpfMOR/ZoQnXxB1woqkUbdLGFDgauo00Py9IudiaqSeiBiulSV9i6XIPig==", + "dependencies": { + "@types/node": "*" + } + }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dependencies": { + "@types/node": "*" + } + }, + "node_modules/abort-controller": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/abort-controller/-/abort-controller-3.0.0.tgz", + "integrity": "sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg==", + "dependencies": { + "event-target-shim": "^5.0.0" + }, + "engines": { + "node": ">=6.5" + } + }, "node_modules/abstract-logging": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/abstract-logging/-/abstract-logging-2.0.1.tgz", @@ -1044,6 +1080,75 @@ "node": ">= 6.0.0" } }, + "node_modules/base64-js": { + "version": "1.5.1", + "resolved": "https://registry.npmjs.org/base64-js/-/base64-js-1.5.1.tgz", + "integrity": "sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ] + }, + "node_modules/bl": { + "version": "6.1.6", + "resolved": "https://registry.npmjs.org/bl/-/bl-6.1.6.tgz", + "integrity": "sha512-jLsPgN/YSvPUg9UX0Kd73CXpm2Psg9FxMeCSXnk3WBO3CMT10JMwijubhGfHCnFu6TPn1ei3b975dxv7K2pWVg==", + "dependencies": { + "@types/readable-stream": "^4.0.0", + "buffer": "^6.0.3", + "inherits": "^2.0.4", + "readable-stream": "^4.2.0" + } + }, + "node_modules/broker-factory": { + "version": "3.1.15", + "resolved": "https://registry.npmjs.org/broker-factory/-/broker-factory-3.1.15.tgz", + "integrity": "sha512-ko+aWvgNuP49meGrdjUu7rC+Y+Wai3cCPxP3xWwHsHfehFjOh5ZQM2yC4gEB2UddeZ/YXhm0K1eG/L6fxym2Og==", + "dependencies": { + "@babel/runtime": "^7.29.7", + "fast-unique-numbers": "^9.0.27", + "tslib": "^2.8.1", + "worker-factory": "^7.0.50" + } + }, + "node_modules/buffer": { + "version": "6.0.3", + "resolved": "https://registry.npmjs.org/buffer/-/buffer-6.0.3.tgz", + "integrity": "sha512-FTiCpNxtwiZZHEZbcbTIcZjERVICn9yq/pDFkTl95/AxzD1naBctN7YO68riM/gLSDY7sdrMby8hofADYuuqOA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ], + "dependencies": { + "base64-js": "^1.3.1", + "ieee754": "^1.2.1" + } + }, + "node_modules/buffer-from": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/buffer-from/-/buffer-from-1.1.2.tgz", + "integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==" + }, "node_modules/color": { "version": "4.2.3", "resolved": "https://registry.npmjs.org/color/-/color-4.2.3.tgz", @@ -1081,6 +1186,38 @@ "simple-swizzle": "^0.2.2" } }, + "node_modules/commist": { + "version": "3.2.0", + "resolved": "https://registry.npmjs.org/commist/-/commist-3.2.0.tgz", + "integrity": "sha512-4PIMoPniho+LqXmpS5d3NuGYncG6XWlkBSVGiWycL22dd42OYdUGil2CWuzklaJoNxyxUSpO4MKIBU94viWNAw==" + }, + "node_modules/concat-stream": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/concat-stream/-/concat-stream-2.0.0.tgz", + "integrity": "sha512-MWufYdFw53ccGjCA+Ol7XJYpAlW6/prSMzuPOTRnJGcGzuhLn4Scrz7qf6o8bROZ514ltazcIFJZevcfbo0x7A==", + "engines": [ + "node >= 6.0" + ], + "dependencies": { + "buffer-from": "^1.0.0", + "inherits": "^2.0.3", + "readable-stream": "^3.0.2", + "typedarray": "^0.0.6" + } + }, + "node_modules/concat-stream/node_modules/readable-stream": { + "version": "3.6.2", + "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz", + "integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==", + "dependencies": { + "inherits": "^2.0.3", + "string_decoder": "^1.1.1", + "util-deprecate": "^1.0.1" + }, + "engines": { + "node": ">= 6" + } + }, "node_modules/cookie": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/cookie/-/cookie-1.1.1.tgz", @@ -1093,6 +1230,22 @@ "url": "https://opencollective.com/express" } }, + "node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, "node_modules/denque": { "version": "2.1.0", "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz", @@ -1158,6 +1311,22 @@ "@esbuild/win32-x64": "0.28.1" } }, + "node_modules/event-target-shim": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/event-target-shim/-/event-target-shim-5.0.1.tgz", + "integrity": "sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ==", + "engines": { + "node": ">=6" + } + }, + "node_modules/events": { + "version": "3.3.0", + "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz", + "integrity": "sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==", + "engines": { + "node": ">=0.8.x" + } + }, "node_modules/fast-decode-uri-component": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/fast-decode-uri-component/-/fast-decode-uri-component-1.0.1.tgz", @@ -1199,6 +1368,18 @@ "fast-decode-uri-component": "^1.0.1" } }, + "node_modules/fast-unique-numbers": { + "version": "9.0.27", + "resolved": "https://registry.npmjs.org/fast-unique-numbers/-/fast-unique-numbers-9.0.27.tgz", + "integrity": "sha512-nDA9ADeINN8SA2u2wCtU+siWFTTDqQR37XvgPIDDmboWQeExz7X0mImxuaN+kJddliIqy2FpVRmnvRZ+j8i1/A==", + "dependencies": { + "@babel/runtime": "^7.29.2", + "tslib": "^2.8.1" + }, + "engines": { + "node": ">=18.2.0" + } + }, "node_modules/fast-uri": { "version": "3.1.2", "resolved": "https://registry.npmjs.org/fast-uri/-/fast-uri-3.1.2.tgz", @@ -1304,6 +1485,11 @@ "is-property": "^1.0.2" } }, + "node_modules/help-me": { + "version": "5.0.0", + "resolved": "https://registry.npmjs.org/help-me/-/help-me-5.0.0.tgz", + "integrity": "sha512-7xgomUX6ADmcYzFik0HzAxh/73YlKR9bmFzf51CZwR+b6YtzU2m0u49hQCqV6SvlqIqsaxovfwdvbnsw3b/zpg==" + }, "node_modules/iconv-lite": { "version": "0.7.2", "resolved": "https://registry.npmjs.org/iconv-lite/-/iconv-lite-0.7.2.tgz", @@ -1319,6 +1505,38 @@ "url": "https://opencollective.com/express" } }, + "node_modules/ieee754": { + "version": "1.2.1", + "resolved": "https://registry.npmjs.org/ieee754/-/ieee754-1.2.1.tgz", + "integrity": "sha512-dcyqhDvX1C46lXZcVqCpK+FtMRQVdIMN6/Df5js2zouUsqG7I6sFxitIC+7KYK29KdXOLHdu9zL4sFnoVQnqaA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ] + }, + "node_modules/inherits": { + "version": "2.0.4", + "resolved": "https://registry.npmjs.org/inherits/-/inherits-2.0.4.tgz", + "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==" + }, + "node_modules/ip-address": { + "version": "10.2.0", + "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.2.0.tgz", + "integrity": "sha512-/+S6j4E9AHvW9SWMSEY9Xfy66O5PWvVEJ08O0y5JGyEKQpojb0K0GKpz/v5HJ/G0vi3D2sjGK78119oXZeE0qA==", + "engines": { + "node": ">= 12" + } + }, "node_modules/ipaddr.js": { "version": "2.4.0", "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-2.4.0.tgz", @@ -1337,6 +1555,15 @@ "resolved": "https://registry.npmjs.org/is-property/-/is-property-1.0.2.tgz", "integrity": "sha512-Ks/IoX00TtClbGQr4TWXemAnktAQvYB7HzcCxDGqEZU6oCmb2INHuOoKxbtR+HFkmYWBKv/dOZtGRiAjDhj92g==" }, + "node_modules/js-sdsl": { + "version": "4.3.0", + "resolved": "https://registry.npmjs.org/js-sdsl/-/js-sdsl-4.3.0.tgz", + "integrity": "sha512-mifzlm2+5nZ+lEcLJMoBK0/IH/bDg8XnJfd/Wq6IP+xoCjLZsTOnV2QpxlVbX9bMnkl5PdEjNtBJ9Cj1NjifhQ==", + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/js-sdsl" + } + }, "node_modules/json-schema-ref-resolver": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/json-schema-ref-resolver/-/json-schema-ref-resolver-3.0.0.tgz", @@ -1400,6 +1627,11 @@ "resolved": "https://registry.npmjs.org/long/-/long-5.3.2.tgz", "integrity": "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA==" }, + "node_modules/lru-cache": { + "version": "10.4.3", + "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-10.4.3.tgz", + "integrity": "sha512-JNAzZcXrCt42VGLuYz0zfAzDfAvJWW6AfYlDBQyDV5DClI2m5sAmK+OIO7s59XfsRsWHp02jAJrRadPRGTt6SQ==" + }, "node_modules/lru.min": { "version": "1.1.4", "resolved": "https://registry.npmjs.org/lru.min/-/lru.min-1.1.4.tgz", @@ -1414,6 +1646,60 @@ "url": "https://github.com/sponsors/wellwelwel" } }, + "node_modules/minimist": { + "version": "1.2.8", + "resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.8.tgz", + "integrity": "sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==", + "funding": { + "url": "https://github.com/sponsors/ljharb" + } + }, + "node_modules/mqtt": { + "version": "5.15.1", + "resolved": "https://registry.npmjs.org/mqtt/-/mqtt-5.15.1.tgz", + "integrity": "sha512-V1WnkGuJh3ec9QXzy5Iylw8OOBK+Xu1WhxcQ9mMpLThG+/JZIMV1PgLNRgIiqXhZnvnVLsuyxHl5A/3bHHbcAA==", + "dependencies": { + "@types/readable-stream": "^4.0.21", + "@types/ws": "^8.18.1", + "commist": "^3.2.0", + "concat-stream": "^2.0.0", + "debug": "^4.4.1", + "help-me": "^5.0.0", + "lru-cache": "^10.4.3", + "minimist": "^1.2.8", + "mqtt-packet": "^9.0.2", + "number-allocator": "^1.0.14", + "readable-stream": "^4.7.0", + "rfdc": "^1.4.1", + "socks": "^2.8.6", + "split2": "^4.2.0", + "worker-timers": "^8.0.23", + "ws": "^8.18.3" + }, + "bin": { + "mqtt": "build/bin/mqtt.js", + "mqtt_pub": "build/bin/pub.js", + "mqtt_sub": "build/bin/sub.js" + }, + "engines": { + "node": ">=16.0.0" + } + }, + "node_modules/mqtt-packet": { + "version": "9.0.2", + "resolved": "https://registry.npmjs.org/mqtt-packet/-/mqtt-packet-9.0.2.tgz", + "integrity": "sha512-MvIY0B8/qjq7bKxdN1eD+nrljoeaai+qjLJgfRn3TiMuz0pamsIWY2bFODPZMSNmabsLANXsLl4EMoWvlaTZWA==", + "dependencies": { + "bl": "^6.0.8", + "debug": "^4.3.4", + "process-nextick-args": "^2.0.1" + } + }, + "node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==" + }, "node_modules/mysql2": { "version": "3.22.5", "resolved": "https://registry.npmjs.org/mysql2/-/mysql2-3.22.5.tgz", @@ -1446,6 +1732,15 @@ "node": ">=8.0.0" } }, + "node_modules/number-allocator": { + "version": "1.0.14", + "resolved": "https://registry.npmjs.org/number-allocator/-/number-allocator-1.0.14.tgz", + "integrity": "sha512-OrL44UTVAvkKdOdRQZIJpLkAdjXGTRda052sN4sO77bKEzYYqWKMBjQvrJFzqygI99gL6Z4u2xctPW1tB8ErvA==", + "dependencies": { + "debug": "^4.3.1", + "js-sdsl": "4.3.0" + } + }, "node_modules/on-exit-leak-free": { "version": "2.1.2", "resolved": "https://registry.npmjs.org/on-exit-leak-free/-/on-exit-leak-free-2.1.2.tgz", @@ -1488,6 +1783,19 @@ "resolved": "https://registry.npmjs.org/pino-std-serializers/-/pino-std-serializers-7.1.0.tgz", "integrity": "sha512-BndPH67/JxGExRgiX1dX0w1FvZck5Wa4aal9198SrRhZjH3GxKQUKIBnYJTdj2HDN3UQAS06HlfcSbQj2OHmaw==" }, + "node_modules/process": { + "version": "0.11.10", + "resolved": "https://registry.npmjs.org/process/-/process-0.11.10.tgz", + "integrity": "sha512-cdGef/drWFoydD1JsMzuFf8100nZl+GT+yacc2bEced5f9Rjk4z+WtFUTBu9PhOi9j/jfmBPu0mMEY4wIdAF8A==", + "engines": { + "node": ">= 0.6.0" + } + }, + "node_modules/process-nextick-args": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/process-nextick-args/-/process-nextick-args-2.0.1.tgz", + "integrity": "sha512-3ouUOpQhtgrbOa17J7+uxOTpITYWaGP7/AhoR3+A+/1e9skrzelGi/dXzEYyvbxubEF6Wn2ypscTKiKJFFn1ag==" + }, "node_modules/process-warning": { "version": "5.0.0", "resolved": "https://registry.npmjs.org/process-warning/-/process-warning-5.0.0.tgz", @@ -1508,6 +1816,21 @@ "resolved": "https://registry.npmjs.org/quick-format-unescaped/-/quick-format-unescaped-4.0.4.tgz", "integrity": "sha512-tYC1Q1hgyRuHgloV/YXs2w15unPVh8qfu/qCTfhTYamaw7fyhumKa2yGpdSo87vY32rIclj+4fWYQXUMs9EHvg==" }, + "node_modules/readable-stream": { + "version": "4.7.0", + "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-4.7.0.tgz", + "integrity": "sha512-oIGGmcpTLwPga8Bn6/Z75SVaH1z5dUut2ibSyAMVhmUggWpmDn2dapB0n7f8nwaSiRtepAsfJyfXIO5DCVAODg==", + "dependencies": { + "abort-controller": "^3.0.0", + "buffer": "^6.0.3", + "events": "^3.3.0", + "process": "^0.11.10", + "string_decoder": "^1.3.0" + }, + "engines": { + "node": "^12.22.0 || ^14.17.0 || >=16.0.0" + } + }, "node_modules/real-require": { "version": "0.2.0", "resolved": "https://registry.npmjs.org/real-require/-/real-require-0.2.0.tgz", @@ -1546,6 +1869,25 @@ "resolved": "https://registry.npmjs.org/rfdc/-/rfdc-1.4.1.tgz", "integrity": "sha512-q1b3N5QkRUWUl7iyylaaj3kOpIT0N2i9MqIEQXP73GVsN9cw3fdx8X63cEmWhJGi2PPCF23Ijp7ktmd39rawIA==" }, + "node_modules/safe-buffer": { + "version": "5.2.1", + "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.2.1.tgz", + "integrity": "sha512-rp3So07KcdmmKbGvgaNxQSJr7bGVSVk5S9Eq1F+ppbRo70+YeaDxkw5Dd8NPN+GD6bjnYm2VuPuCXmpuYvmCXQ==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ] + }, "node_modules/safe-regex2": { "version": "5.1.1", "resolved": "https://registry.npmjs.org/safe-regex2/-/safe-regex2-5.1.1.tgz", @@ -1658,6 +2000,28 @@ "is-arrayish": "^0.3.1" } }, + "node_modules/smart-buffer": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/smart-buffer/-/smart-buffer-4.2.0.tgz", + "integrity": "sha512-94hK0Hh8rPqQl2xXc3HsaBoOXKV20MToPkcXvwbISWLEs+64sBq5kFgn2kJDHb1Pry9yrP0dxrCI9RRci7RXKg==", + "engines": { + "node": ">= 6.0.0", + "npm": ">= 3.0.0" + } + }, + "node_modules/socks": { + "version": "2.8.9", + "resolved": "https://registry.npmjs.org/socks/-/socks-2.8.9.tgz", + "integrity": "sha512-LJhUYUvItdQ0LkJTmPeaEObWXAqFyfmP85x0tch/ez9cahmhlBBLbIqDFnvBnUJGagb0JbIQrkBs1wJ+yRYpEw==", + "dependencies": { + "ip-address": "^10.1.1", + "smart-buffer": "^4.2.0" + }, + "engines": { + "node": ">= 10.0.0", + "npm": ">= 3.0.0" + } + }, "node_modules/sonic-boom": { "version": "4.2.1", "resolved": "https://registry.npmjs.org/sonic-boom/-/sonic-boom-4.2.1.tgz", @@ -1688,6 +2052,14 @@ "url": "https://github.com/mysqljs/sql-escaper?sponsor=1" } }, + "node_modules/string_decoder": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz", + "integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==", + "dependencies": { + "safe-buffer": "~5.2.0" + } + }, "node_modules/thread-stream": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/thread-stream/-/thread-stream-3.2.0.tgz", @@ -1707,8 +2079,7 @@ "node_modules/tslib": { "version": "2.8.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", - "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", - "optional": true + "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==" }, "node_modules/tsx": { "version": "4.22.4", @@ -1728,6 +2099,11 @@ "fsevents": "~2.3.3" } }, + "node_modules/typedarray": { + "version": "0.0.6", + "resolved": "https://registry.npmjs.org/typedarray/-/typedarray-0.0.6.tgz", + "integrity": "sha512-/aCDEGatGvZ2BIk+HmLf4ifCJFwvKFNb9/JeZPMulfgFracn9QFcAf5GO8B/mweUjSoblS5In0cWhqpfs/5PQA==" + }, "node_modules/typescript": { "version": "5.9.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", @@ -1746,6 +2122,74 @@ "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz", "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==" }, + "node_modules/util-deprecate": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz", + "integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==" + }, + "node_modules/worker-factory": { + "version": "7.0.50", + "resolved": "https://registry.npmjs.org/worker-factory/-/worker-factory-7.0.50.tgz", + "integrity": "sha512-hhwc0G+sFwM4qBuhJIUBn2p1Jf8v/FwmLUANBf/Q+Lt2uI8mfIZQhXaZQACodQD4R7Zp6cn/6702bIvNn2puJQ==", + "dependencies": { + "@babel/runtime": "^7.29.7", + "fast-unique-numbers": "^9.0.27", + "tslib": "^2.8.1" + } + }, + "node_modules/worker-timers": { + "version": "8.0.32", + "resolved": "https://registry.npmjs.org/worker-timers/-/worker-timers-8.0.32.tgz", + "integrity": "sha512-huEv5mB6xIqcadsZ8SuuFQH9Lg9Hq0h0xD9vAZZsLPNw0aLxoDWyjTALAUD3hAA4cuKbKaqrjbBcbEYClwSh3w==", + "dependencies": { + "@babel/runtime": "^7.29.7", + "tslib": "^2.8.1", + "worker-timers-broker": "^8.0.16", + "worker-timers-worker": "^9.0.14" + } + }, + "node_modules/worker-timers-broker": { + "version": "8.0.17", + "resolved": "https://registry.npmjs.org/worker-timers-broker/-/worker-timers-broker-8.0.17.tgz", + "integrity": "sha512-avGDVB9AP5k5eCAP2AiT/nwCHGCNla7Z+nMZWXpmhbiSTry6A7VwEzlffTO34/5Eo55O+XLbWQu0bBW3uEO0UA==", + "dependencies": { + "@babel/runtime": "^7.29.7", + "broker-factory": "^3.1.15", + "fast-unique-numbers": "^9.0.27", + "tslib": "^2.8.1", + "worker-timers-worker": "^9.0.14" + } + }, + "node_modules/worker-timers-worker": { + "version": "9.0.15", + "resolved": "https://registry.npmjs.org/worker-timers-worker/-/worker-timers-worker-9.0.15.tgz", + "integrity": "sha512-KKUe7lZ/Aignr51H6hOUik8LwTnIgojH/1lwhli8A8qIEIyewogZTpNpMW5B6BF7nmwBOkUoTYgf1H3QShcjSA==", + "dependencies": { + "@babel/runtime": "^7.29.7", + "tslib": "^2.8.1", + "worker-factory": "^7.0.50" + } + }, + "node_modules/ws": { + "version": "8.21.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", + "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/zod": { "version": "3.25.76", "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", diff --git a/backend/package.json b/backend/package.json index 617f2c6..f2250df 100644 --- a/backend/package.json +++ b/backend/package.json @@ -17,12 +17,13 @@ "db:migrate:verify": "npm run build && node dist/db/migrate-cli.js verify", "db:migrate:down": "npm run build && node dist/db/migrate-cli.js down", "test:mysql:migration": "npm run build && node tests/mysql-migration-roundtrip.test.mjs", - "test": "npm run build && node tests/backend-contract.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs && node tests/task-repository.test.mjs && node tests/platform-config-repository.test.mjs && node tests/auth.test.mjs && node tests/rbac.test.mjs && node tests/user-management.test.mjs && node tests/store-room.test.mjs && node tests/content-management.test.mjs && node tests/store-discovery.test.mjs && node tests/store-access.test.mjs && node tests/pricing.test.mjs && node tests/order-state.test.mjs && node tests/order-management.test.mjs && node tests/order-share.test.mjs && node tests/payment.test.mjs && node tests/wechat-pay.test.mjs && node tests/third-party.test.mjs && node tests/profit-sharing.test.mjs" + "test": "npm run build && node tests/backend-contract.test.mjs && node tests/mqtt-service.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs && node tests/task-repository.test.mjs && node tests/platform-config-repository.test.mjs && node tests/auth.test.mjs && node tests/rbac.test.mjs && node tests/user-management.test.mjs && node tests/store-room.test.mjs && node tests/content-management.test.mjs && node tests/store-discovery.test.mjs && node tests/store-access.test.mjs && node tests/pricing.test.mjs && node tests/order-state.test.mjs && node tests/order-management.test.mjs && node tests/order-share.test.mjs && node tests/payment.test.mjs && node tests/wechat-pay.test.mjs && node tests/third-party.test.mjs && node tests/profit-sharing.test.mjs" }, "dependencies": { "@fastify/cors": "^11.2.0", "@fastify/rate-limit": "^11.0.0", "fastify": "^5.8.5", + "mqtt": "^5.15.1", "mysql2": "^3.11.3", "pino": "^9.4.0", "sharp": "^0.34.0", diff --git a/backend/src/app.ts b/backend/src/app.ts index 9f20d97..0bdefac 100644 --- a/backend/src/app.ts +++ b/backend/src/app.ts @@ -3,7 +3,7 @@ import cors from '@fastify/cors'; import rateLimit from '@fastify/rate-limit'; import Fastify, { type FastifyInstance } from 'fastify'; import { loadConfig, type AppConfig } from './config.js'; -import { registerHealthRoutes } from './routes/health.js'; +import { registerHealthRoutes, type MqttHealthProvider } from './routes/health.js'; import { registerPlatformBootstrapRoutes, type PlatformConfigResolver @@ -57,6 +57,7 @@ export interface BuildAppOptions { orderShare?: OrderShareRouteOptions; payment?: PaymentRouteOptions; thirdParty?: ThirdPartyRouteOptions; + mqtt?: MqttHealthProvider; } declare module 'fastify' { @@ -100,7 +101,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise; @@ -35,6 +39,14 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env) { ) { throw new Error('QIPAI_JWT_SECRET must be explicitly configured in production.'); } + const mqttUsernameConfigured = parsed.QIPAI_MQTT_USERNAME.length > 0; + const mqttPasswordConfigured = parsed.QIPAI_MQTT_PASSWORD.length > 0; + if (mqttUsernameConfigured !== mqttPasswordConfigured) { + throw new Error('QIPAI_MQTT_USERNAME and QIPAI_MQTT_PASSWORD must be configured together.'); + } + if (parsed.NODE_ENV === 'production' && !mqttUsernameConfigured) { + throw new Error('MQTT credentials must be explicitly configured in production.'); + } return { nodeEnv: parsed.NODE_ENV, @@ -69,8 +81,14 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env) { }, mqtt: { url: parsed.QIPAI_MQTT_URL, - usernameConfigured: parsed.QIPAI_MQTT_USERNAME.length > 0, - passwordConfigured: parsed.QIPAI_MQTT_PASSWORD.length > 0 + clientId: parsed.QIPAI_MQTT_CLIENT_ID, + username: parsed.QIPAI_MQTT_USERNAME, + credential: parsed.QIPAI_MQTT_PASSWORD, + usernameConfigured: mqttUsernameConfigured, + passwordConfigured: mqttPasswordConfigured, + reconnectPeriodMs: parsed.QIPAI_MQTT_RECONNECT_MS, + connectTimeoutMs: parsed.QIPAI_MQTT_CONNECT_TIMEOUT_MS, + maxMessageBytes: parsed.QIPAI_MQTT_MAX_MESSAGE_BYTES } }; } diff --git a/backend/src/mqtt/mqtt-service.ts b/backend/src/mqtt/mqtt-service.ts new file mode 100644 index 0000000..8a97b6d --- /dev/null +++ b/backend/src/mqtt/mqtt-service.ts @@ -0,0 +1,171 @@ +import { connect, type IClientOptions, type IClientPublishOptions } from 'mqtt'; +import type { AppConfig } from '../config.js'; + +export const DEVICE_UPLINK_TOPIC = '/devicesend/+'; +export const DEVICE_WILL_TOPIC = '/devicewill/+'; +const DEVICE_COMMAND_PREFIX = '/deviceaccept/'; +const MQTT_PASSWORD_OPTION = 'password'; + +type MqttEvent = 'connect' | 'reconnect' | 'close' | 'offline' | 'error' | 'message'; + +export interface MqttClientLike { + connected: boolean; + on(event: MqttEvent, listener: (...args: any[]) => void): this; + subscribe( + topics: string[], + options: { qos: 1 }, + callback: (error?: Error | null) => void + ): void; + publish( + topic: string, + payload: Buffer, + options: IClientPublishOptions, + callback: (error?: Error) => void + ): void; + end(force: boolean, options: Record, callback: () => void): void; +} + +export type MqttClientFactory = (url: string, options: IClientOptions) => MqttClientLike; + +export interface MqttHealthSnapshot { + configured: boolean; + connected: boolean; + subscriptionsReady: boolean; + reconnectCount: number; + receivedMessages: number; + rejectedOversizeMessages: number; + lastConnectedAt: string | null; + lastMessageAt: string | null; + lastError: string | null; +} + +export class MqttService { + private client: MqttClientLike | null = null; + private subscriptionsReady = false; + private reconnectCount = 0; + private receivedMessages = 0; + private rejectedOversizeMessages = 0; + private lastConnectedAt: string | null = null; + private lastMessageAt: string | null = null; + private lastError: string | null = null; + + constructor( + private readonly config: AppConfig['mqtt'], + private readonly clientFactory: MqttClientFactory = connect as MqttClientFactory + ) {} + + start(): void { + if (this.client || !this.isConfigured()) { + return; + } + + this.client = this.clientFactory(this.config.url, { + clientId: this.config.clientId, + username: this.config.username, + [MQTT_PASSWORD_OPTION]: this.config.credential, + protocolVersion: 3, + clean: false, + reconnectPeriod: this.config.reconnectPeriodMs, + connectTimeout: this.config.connectTimeoutMs, + resubscribe: false, + queueQoSZero: false + }); + + this.client.on('connect', () => { + this.lastConnectedAt = new Date().toISOString(); + this.lastError = null; + this.subscribeToDeviceTopics(); + }); + this.client.on('reconnect', () => { + this.reconnectCount += 1; + this.subscriptionsReady = false; + }); + this.client.on('close', () => { + this.subscriptionsReady = false; + }); + this.client.on('offline', () => { + this.subscriptionsReady = false; + }); + this.client.on('error', (error: Error) => { + this.lastError = sanitizeError(error); + }); + this.client.on('message', (_topic: string, payload: Buffer) => { + if (payload.length > this.config.maxMessageBytes) { + this.rejectedOversizeMessages += 1; + this.lastError = `MQTT message exceeded ${this.config.maxMessageBytes} bytes`; + return; + } + this.receivedMessages += 1; + this.lastMessageAt = new Date().toISOString(); + }); + } + + async stop(): Promise { + const client = this.client; + this.client = null; + this.subscriptionsReady = false; + if (!client) { + return; + } + await new Promise((resolve) => client.end(false, {}, resolve)); + } + + async publishDeviceCommand(deviceId: string, payload: Buffer | string): Promise { + if (!/^[A-Za-z0-9_-]{1,64}$/.test(deviceId)) { + throw new Error('Invalid MQTT DeviceID.'); + } + const body = Buffer.isBuffer(payload) ? payload : Buffer.from(payload, 'utf8'); + if (body.length > this.config.maxMessageBytes) { + throw new Error(`MQTT command exceeds ${this.config.maxMessageBytes} bytes.`); + } + if (!this.client?.connected || !this.subscriptionsReady) { + throw new Error('MQTT transport is not ready.'); + } + + await new Promise((resolve, reject) => { + this.client?.publish( + `${DEVICE_COMMAND_PREFIX}${deviceId}`, + body, + { qos: 1, retain: false }, + (error?: Error) => error ? reject(error) : resolve() + ); + }); + } + + health(): MqttHealthSnapshot { + return { + configured: this.isConfigured(), + connected: this.client?.connected === true, + subscriptionsReady: this.subscriptionsReady, + reconnectCount: this.reconnectCount, + receivedMessages: this.receivedMessages, + rejectedOversizeMessages: this.rejectedOversizeMessages, + lastConnectedAt: this.lastConnectedAt, + lastMessageAt: this.lastMessageAt, + lastError: this.lastError + }; + } + + private isConfigured(): boolean { + return this.config.usernameConfigured && this.config.passwordConfigured; + } + + private subscribeToDeviceTopics(): void { + this.subscriptionsReady = false; + this.client?.subscribe( + [DEVICE_UPLINK_TOPIC, DEVICE_WILL_TOPIC], + { qos: 1 }, + (error?: Error | null) => { + if (error) { + this.lastError = sanitizeError(error); + return; + } + this.subscriptionsReady = true; + } + ); + } +} + +function sanitizeError(error: Error): string { + return error.message.replace(/(password|username|credential)=\S+/gi, '$1='); +} diff --git a/backend/src/routes/health.ts b/backend/src/routes/health.ts index a203767..0423973 100644 --- a/backend/src/routes/health.ts +++ b/backend/src/routes/health.ts @@ -12,10 +12,24 @@ interface ReadyPayload extends HealthPayload { checks: { mysqlConfigured: boolean; mqttConfigured: boolean; + mqttConnected: boolean; + mqttSubscriptionsReady: boolean; }; } -export async function registerHealthRoutes(app: FastifyInstance, config: AppConfig): Promise { +export interface MqttHealthProvider { + health(): { + configured: boolean; + connected: boolean; + subscriptionsReady: boolean; + }; +} + +export async function registerHealthRoutes( + app: FastifyInstance, + config: AppConfig, + mqtt?: MqttHealthProvider +): Promise { const health = async (request: { traceId: string }): Promise => ({ ok: true, service: 'qipai-api', @@ -23,13 +37,19 @@ export async function registerHealthRoutes(app: FastifyInstance, config: AppConf traceId: request.traceId }); - const ready = async (request: { traceId: string }): Promise => ({ - ...(await health(request)), - checks: { - mysqlConfigured: config.mysql.passwordConfigured, - mqttConfigured: config.mqtt.usernameConfigured && config.mqtt.passwordConfigured - } - }); + const ready = async (request: { traceId: string }): Promise => { + const mqttHealth = mqtt?.health(); + return { + ...(await health(request)), + checks: { + mysqlConfigured: config.mysql.passwordConfigured, + mqttConfigured: mqttHealth?.configured + ?? (config.mqtt.usernameConfigured && config.mqtt.passwordConfigured), + mqttConnected: mqttHealth?.connected ?? false, + mqttSubscriptionsReady: mqttHealth?.subscriptionsReady ?? false + } + }; + }; app.get('/app-api/health', health); app.get('/admin-api/health', health); diff --git a/backend/src/server.ts b/backend/src/server.ts index 9be16b4..f184632 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -26,6 +26,7 @@ import { FetchThirdPartyTransport, parseThirdPartyCredentials, ThirdPartyClient } from './third-party/third-party-client.js'; import { ThirdPartyService } from './third-party/third-party-service.js'; +import { MqttService } from './mqtt/mqtt-service.js'; const config = loadConfig(); const pool = createMySqlPool(config); @@ -36,8 +37,10 @@ const paymentRepository = new PaymentRepository(pool); const wechatCredentials = parseWechatPayCredentials(config.payment.wechatCredentialsJson); const wechatPayClient = new WechatPayClient(new FetchWechatPayTransport()); const thirdPartyCredentials = parseThirdPartyCredentials(config.thirdParty.credentialsJson); +const mqtt = new MqttService(config.mqtt); const app = await buildApp({ config, + mqtt, platformConfigRepository: new PlatformConfigRepository(pool), auth: { repository: authRepository, @@ -133,10 +136,12 @@ const app = await buildApp({ } }); app.addHook('onClose', async () => { + await mqtt.stop(); await closeMySqlPool(pool); }); try { + mqtt.start(); await app.listen({ host: config.host, port: config.port }); } catch (error) { app.log.error(error); diff --git a/backend/tests/mqtt-service.test.mjs b/backend/tests/mqtt-service.test.mjs new file mode 100644 index 0000000..489b3ec --- /dev/null +++ b/backend/tests/mqtt-service.test.mjs @@ -0,0 +1,113 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import { loadConfig } from '../dist/config.js'; +import { buildApp } from '../dist/app.js'; +import { + DEVICE_UPLINK_TOPIC, + DEVICE_WILL_TOPIC, + MqttService +} from '../dist/mqtt/mqtt-service.js'; + +class FakeClient extends EventEmitter { + connected = false; + subscriptions = []; + publications = []; + ended = false; + + subscribe(topics, options, callback) { + this.subscriptions.push({ topics, options }); + callback(); + } + + publish(topic, payload, options, callback) { + this.publications.push({ topic, payload, options }); + callback(); + } + + end(_force, _options, callback) { + this.ended = true; + callback(); + } +} + +const config = loadConfig({ + NODE_ENV: 'test', + QIPAI_MQTT_URL: 'mqtt://127.0.0.1:1883', + QIPAI_MQTT_CLIENT_ID: 'qipai-backend-test', + QIPAI_MQTT_USERNAME: 'backend-test', + QIPAI_MQTT_PASSWORD: 'not-a-real-secret', + QIPAI_MQTT_MAX_MESSAGE_BYTES: '1024' +}); +const fake = new FakeClient(); +let connectOptions; +const service = new MqttService(config.mqtt, (_url, options) => { + connectOptions = options; + return fake; +}); + +service.start(); +assert.equal(connectOptions.protocolVersion, 3); +assert.equal(connectOptions.clean, false); +assert.equal(connectOptions.reconnectPeriod, 3000); +assert.equal(connectOptions.resubscribe, false); +assert.equal(connectOptions.queueQoSZero, false); + +fake.connected = true; +fake.emit('connect'); +assert.deepEqual(fake.subscriptions, [{ + topics: [DEVICE_UPLINK_TOPIC, DEVICE_WILL_TOPIC], + options: { qos: 1 } +}]); +assert.equal(service.health().subscriptionsReady, true); + +await service.publishDeviceCommand('BOX_001', '{"command":"GetInfo"}'); +assert.equal(fake.publications[0].topic, '/deviceaccept/BOX_001'); +assert.equal(fake.publications[0].options.qos, 1); +assert.equal(fake.publications[0].options.retain, false); + +fake.emit('message', '/devicesend/BOX_001', Buffer.from('{"result":"ok"}')); +assert.equal(service.health().receivedMessages, 1); +fake.emit('message', '/devicesend/BOX_001', Buffer.alloc(1025)); +assert.equal(service.health().rejectedOversizeMessages, 1); +assert.match(service.health().lastError, /exceeded 1024 bytes/); + +fake.emit('reconnect'); +assert.equal(service.health().reconnectCount, 1); +assert.equal(service.health().subscriptionsReady, false); +fake.emit('connect'); +assert.equal(fake.subscriptions.length, 2); +assert.equal(service.health().subscriptionsReady, true); + +const app = await buildApp({ config, mqtt: service }); +const ready = await app.inject({ method: 'GET', url: '/app-api/ready' }); +assert.equal(ready.statusCode, 200); +assert.deepEqual(ready.json().checks, { + mysqlConfigured: false, + mqttConfigured: true, + mqttConnected: true, + mqttSubscriptionsReady: true +}); +await app.close(); + +await assert.rejects( + service.publishDeviceCommand('../bad', '{}'), + /Invalid MQTT DeviceID/ +); +await assert.rejects( + service.publishDeviceCommand('BOX_001', Buffer.alloc(1025)), + /exceeds 1024 bytes/ +); + +await service.stop(); +assert.equal(fake.ended, true); +assert.equal(service.health().connected, false); + +assert.throws( + () => loadConfig({ + NODE_ENV: 'test', + QIPAI_MQTT_USERNAME: 'only-user' + }), + /must be configured together/ +); + +console.log('PASS: MQTT 3.1 transport, QoS 1, resubscribe, health and size limits work.'); diff --git a/backend/tests/payment.test.mjs b/backend/tests/payment.test.mjs index 98a6636..023e455 100644 --- a/backend/tests/payment.test.mjs +++ b/backend/tests/payment.test.mjs @@ -5,6 +5,8 @@ import { signAccessToken } from '../dist/auth/jwt.js'; assert.equal(loadConfig({ NODE_ENV: 'production', + QIPAI_MQTT_USERNAME: 'backend-test', + QIPAI_MQTT_PASSWORD: 'not-a-real-secret', QIPAI_JWT_SECRET: 'production-test-secret-that-is-long-enough', QIPAI_TEST_PAYMENT_ENABLED: 'true' }).payment.testAdapterEnabled, false); diff --git a/backend/tests/profit-sharing.test.mjs b/backend/tests/profit-sharing.test.mjs index c51eda0..ceb9ecd 100644 --- a/backend/tests/profit-sharing.test.mjs +++ b/backend/tests/profit-sharing.test.mjs @@ -7,6 +7,8 @@ import { WechatPayClient } from '../dist/payments/wechat-pay-client.js'; assert.equal(loadConfig({ NODE_ENV: 'production', + QIPAI_MQTT_USERNAME: 'backend-test', + QIPAI_MQTT_PASSWORD: 'not-a-real-secret', QIPAI_JWT_SECRET: 'production-profit-share-secret-long-enough', QIPAI_PROFIT_SHARE_MOCK_ENABLED: 'true' }).payment.profitShareMockEnabled, false); diff --git a/deploy/emqx/acl.conf.template b/deploy/emqx/acl.conf.template index 74f4e4b..c4a9e62 100644 --- a/deploy/emqx/acl.conf.template +++ b/deploy/emqx/acl.conf.template @@ -1,25 +1,25 @@ -%% Qipai EMQX ACL baseline for device topic isolation. +%% Qipai EMQX ACL for the selected Jilian device protocol. %% EMQX ACL files use Erlang tuple rules and are matched from top to bottom. -%% Replace example usernames and topic namespaces after M06 device onboarding. +%% Device MQTT username must equal its DeviceID. Credentials are created outside Git. -%% Device clients may publish telemetry/status under their own client id. +%% A device may publish only its uplink and will topics. {allow, {clientid, {re, "^qipai-device-[A-Za-z0-9_-]+$"}}, publish, [ - "qipai/${clientid}/telemetry", - "qipai/${clientid}/status", - "qipai/${clientid}/event/#" + "/devicesend/${username}", + "/devicewill/${username}" ]}. -%% Device clients may subscribe only to commands for their own client id. +%% A device may subscribe only to its command topic. {allow, {clientid, {re, "^qipai-device-[A-Za-z0-9_-]+$"}}, subscribe, [ - "qipai/${clientid}/command/#" + "/deviceaccept/${username}" ]}. -%% Backend service may publish commands and subscribe to device events. -{allow, {username, "qipai_backend"}, all, [ - "qipai/+/command/#", - "qipai/+/telemetry", - "qipai/+/status", - "qipai/+/event/#" +%% Backend service subscribes to uplink/will and publishes commands. +{allow, {username, "qipai_backend_prod"}, subscribe, [ + "/devicesend/+", + "/devicewill/+" +]}. +{allow, {username, "qipai_backend_prod"}, publish, [ + "/deviceaccept/+" ]}. %% Do not expose system topics or broad wildcards to ordinary clients. diff --git a/scripts/dev/wsl/mqtt-smoke.sh b/scripts/dev/wsl/mqtt-smoke.sh index 5be3545..0ec7f49 100644 --- a/scripts/dev/wsl/mqtt-smoke.sh +++ b/scripts/dev/wsl/mqtt-smoke.sh @@ -13,9 +13,10 @@ port="${QIPAI_MQTT_PORT:-1883}" protocol="${QIPAI_MQTT_PROTOCOL:-mqtt}" mqtt_version="${QIPAI_MQTT_VERSION:-3.1.1}" qos="${QIPAI_MQTT_QOS:-1}" -client_id="qipai-device-smoke-$(date +%s)" -allowed_publish_topic="${QIPAI_MQTT_ALLOWED_PUBLISH_TOPIC:-qipai/${client_id}/status}" -allowed_subscribe_topic="${QIPAI_MQTT_ALLOWED_SUBSCRIBE_TOPIC:-qipai/${client_id}/command/#}" +device_id="${QIPAI_MQTT_DEVICE_ID:-SMOKE_DEVICE}" +client_id="qipai-device-${device_id}" +allowed_publish_topic="${QIPAI_MQTT_ALLOWED_PUBLISH_TOPIC:-/devicesend/${device_id}}" +allowed_subscribe_topic="${QIPAI_MQTT_ALLOWED_SUBSCRIBE_TOPIC:-/deviceaccept/${device_id}}" denied_subscribe_topic="${QIPAI_MQTT_DENIED_SUBSCRIBE_TOPIC:-\$SYS/#}" enable_tls_smoke="${QIPAI_MQTT_ENABLE_TLS_SMOKE:-false}" enable_will_smoke="${QIPAI_MQTT_ENABLE_WILL_SMOKE:-false}" @@ -24,9 +25,9 @@ tls_host="${QIPAI_MQTT_TLS_HOST:-$host}" tls_port="${QIPAI_MQTT_TLS_PORT:-8883}" tls_protocol="${QIPAI_MQTT_TLS_PROTOCOL:-mqtts}" tls_insecure="${QIPAI_MQTT_TLS_INSECURE:-false}" -will_topic="${QIPAI_MQTT_WILL_TOPIC:-qipai/${client_id}/will}" +will_topic="${QIPAI_MQTT_WILL_TOPIC:-/devicewill/${device_id}}" will_message="${QIPAI_MQTT_WILL_MESSAGE:-{\"type\":\"will\",\"clientId\":\"${client_id}\"}}" -idempotency_topic="${QIPAI_MQTT_IDEMPOTENCY_TOPIC:-qipai/${client_id}/status}" +idempotency_topic="${QIPAI_MQTT_IDEMPOTENCY_TOPIC:-/devicesend/${device_id}}" idempotency_key="${QIPAI_MQTT_IDEMPOTENCY_KEY:-smoke-${client_id}}" echo "INFO: MQTT smoke target ${protocol}://${host}:${port}, version ${mqtt_version}, qos ${qos}" diff --git a/scripts/setup/emqx.sh b/scripts/setup/emqx.sh index 340eced..a2f19db 100644 --- a/scripts/setup/emqx.sh +++ b/scripts/setup/emqx.sh @@ -5,6 +5,52 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" # shellcheck source=lib.sh . "${SCRIPT_DIR}/lib.sh" +qipai_emqx_install_or_update() { + qipai_require_root_for_write + qipai_check_arch + + if command -v mqttx >/dev/null 2>&1; then + qipai_fail "MQTTX must not be installed on the production server" + return 1 + fi + + export DEBIAN_FRONTEND=noninteractive + apt-get update + apt-get install -y curl ca-certificates gnupg lsb-release + + if ! command -v emqx >/dev/null 2>&1; then + local installer + installer="/tmp/install-emqx-deb.sh" + curl -fsSL https://assets.emqx.com/scripts/install-emqx-deb.sh -o "$installer" + bash -n "$installer" + bash "$installer" + fi + + apt-get update + apt-get install -y emqx + systemctl enable --now emqx + qipai_pass "EMQX package installed/updated and service enabled" + qipai_emqx_status + qipai_info "Authentication users and secrets must be created from /etc/qipai/qipai.secrets; they are never printed here." + qipai_info "Apply deploy/emqx/acl.conf.template only after creating qipai_backend_prod and device credentials." +} + +qipai_emqx_manage() { + qipai_emqx_status + if command -v emqx >/dev/null 2>&1; then + printf "输入 U 更新 EMQX，直接回车仅查看状态: " + else + printf "输入 I 安装 EMQX，直接回车仅查看状态: " + fi + local choice + read -r choice || true + case "${choice}" in + I|i|U|u) qipai_emqx_install_or_update ;; + "") qipai_info "EMQX status check complete; no package changes requested" ;; + *) qipai_warn "unknown choice; no package changes made" ;; + esac +} + qipai_emqx_status() { local acl_template authz_template acl_template="${SCRIPT_DIR}/../../deploy/emqx/acl.conf.template" @@ -20,6 +66,9 @@ qipai_emqx_status() { if command -v emqx >/dev/null 2>&1; then qipai_pass "emqx command exists: $(command -v emqx)" emqx version 2>/dev/null || true + if command -v dpkg-query >/dev/null 2>&1; then + dpkg-query -W -f='${Package} ${Version} ${Architecture}\n' emqx 2>/dev/null || true + fi else qipai_warn "emqx command not found" fi @@ -45,6 +94,12 @@ qipai_emqx_status() { fi done + if command -v mqttx >/dev/null 2>&1; then + qipai_fail "MQTTX is installed on the server; production policy forbids it" + else + qipai_pass "MQTTX is not installed on the server" + fi + [ -f "$acl_template" ] && qipai_pass "ACL template exists" || qipai_warn "ACL template not found" [ -f "$authz_template" ] && qipai_pass "authorization template exists" || qipai_warn "authorization template not found" @@ -54,4 +109,6 @@ qipai_emqx_status() { if [ "${1:-}" = "--run" ]; then qipai_emqx_status +elif [ "${1:-}" = "--install" ]; then + qipai_emqx_install_or_update fi diff --git a/setup.sh b/setup.sh index 3bd63ed..0f6e396 100644 --- a/setup.sh +++ b/setup.sh @@ -62,7 +62,7 @@ show_status() { } show_mqtt_status() { - qipai_emqx_status + qipai_emqx_manage } show_https_status() {