From 5498e2e52f1ed33839d7b571409e3176eee8aa05 Mon Sep 17 00:00:00 2001 From: sunlei Date: Tue, 16 Jun 2026 15:48:29 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8DQQBot=E6=8F=92?= =?UTF-8?q?=E4=BB=B6Worker=E5=B9=B6=E5=8F=91=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 8 + API.md | 4 +- Jenkinsfile | 2 + README.md | 4 +- k8s/prod/api.yaml | 80 +++- package.json | 2 + pnpm-lock.yaml | 223 ++++++++++- .../application/plugin-platform.service.ts | 149 +++++-- .../builtin-plugin-worker-runtime.factory.ts | 38 +- .../bullmq-plugin-worker-request.queue.ts | 318 +++++++++++++++ .../integration/runtime/index.ts | 1 + .../integration/runtime/worker-runtime.ts | 207 +++++++++- .../runtime/worker-runtime.types.ts | 6 + .../plugin-platform/plugin-platform.module.ts | 17 +- .../plugin-lifecycle-runtime.spec.ts | 213 +++++++++- .../plugin-platform/worker-runtime.spec.ts | 367 +++++++++++++++++- 16 files changed, 1551 insertions(+), 88 deletions(-) create mode 100644 src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts diff --git a/.env.example b/.env.example index e8fffc3..3c788b5 100644 --- a/.env.example +++ b/.env.example @@ -55,6 +55,14 @@ QQBOT_SEND_GLOBAL_INTERVAL_MS=2500 QQBOT_SEND_TARGET_INTERVAL_MS=8000 QQBOT_SEND_JITTER_MS=800 QQBOT_SEND_MAX_QUEUE_WAIT_MS=30000 +# K8s production can use QQBOT_PLUGIN_QUEUE_REDIS_HOST=kt-qqbot-plugin-redis. +QQBOT_PLUGIN_QUEUE_REDIS_HOST=127.0.0.1 +QQBOT_PLUGIN_QUEUE_REDIS_PORT=6379 +QQBOT_PLUGIN_QUEUE_REDIS_PASSWORD= +QQBOT_PLUGIN_QUEUE_REDIS_DB=0 +QQBOT_PLUGIN_QUEUE_REDIS_PREFIX=kt:qqbot:plugin-worker +QQBOT_PLUGIN_QUEUE_REMOVE_ON_FAIL=100 +QQBOT_PLUGIN_QUEUE_WAIT_BUFFER_MS=5000 QQBOT_COMMAND_MIN_COOLDOWN_MS=5000 QQBOT_RULE_MIN_COOLDOWN_MS=30000 QQBOT_REPEATER_THRESHOLD=4 diff --git a/API.md b/API.md index 63bcb9c..92cd107 100644 --- a/API.md +++ b/API.md @@ -85,7 +85,7 @@ Admin、Component、Dict、MinIO、Blog 管理、WordPress 管理和 QQBot 管 | Admin | `ADMIN_TOKEN_SECRET`、`ADMIN_COOKIE_SECURE`、`SNOWFLAKE_WORKER_ID`、`SNOWFLAKE_DATACENTER_ID` | | WordPress | `WORDPRESS_BASE_URL`、`WORDPRESS_HOST_HEADER`、`WORDPRESS_ADMIN_USERNAME`、`WORDPRESS_ADMIN_PASSWORD` | | Loki | `LOG_LEVEL`、`LOG_APP_NAME`、`LOKI_URL`、`LOKI_QUERY_HOST`、`LOKI_QUERY_SELECTOR` | -| QQBot | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_PATH`、`QQBOT_REVERSE_WS_TOKEN`、`QQBOT_EVENT_BUS`、`QQBOT_SEND_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*` | +| QQBot | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_PATH`、`QQBOT_REVERSE_WS_TOKEN`、`QQBOT_EVENT_BUS`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*` | | NapCat | `NAPCAT_WEBUI_BASE_URL`、`NAPCAT_WEBUI_TOKEN`、`QQBOT_NAPCAT_*` | | MQTT | `MQTT_URL`、`MQTT_USERNAME`、`MQTT_PASSWORD`、`MQTT_CLIENT_ID` | | BangDream | `BANGDREAM_TSUGU_MAIN_SERVER`、`BANGDREAM_TSUGU_DISPLAYED_SERVERS`、`BANGDREAM_TSUGU_CACHE_ROOT` | @@ -94,6 +94,8 @@ Admin、Component、Dict、MinIO、Blog 管理、WordPress 管理和 QQBot 管 真实密码、Token、OAuth secret 和生产 env 不提交到 Git。 +QQBot 插件 worker 队列依赖 Redis。K8s 生产清单提供内部 Redis Service `kt-qqbot-plugin-redis:6379`，用于 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` / `QQBOT_PLUGIN_QUEUE_REDIS_PORT`。 + ## Admin 与基础后台 ### Auth / User diff --git a/Jenkinsfile b/Jenkinsfile index a80b371..9568cf4 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -36,6 +36,8 @@ def requiredRuntimeEnvKeys() { 'WORDPRESS_BASE_URL', 'FFLOGS_CLIENT_ID', 'FFLOGS_CLIENT_SECRET', + 'QQBOT_PLUGIN_QUEUE_REDIS_HOST', + 'QQBOT_PLUGIN_QUEUE_REDIS_PORT', ] } diff --git a/README.md b/README.md index 4905afa..df829d5 100644 --- a/README.md +++ b/README.md @@ -62,13 +62,15 @@ ci/ Jenkins Agent/Docker 辅助文件 | Admin | `ADMIN_TOKEN_SECRET`、`ADMIN_COOKIE_SECURE`、`SNOWFLAKE_WORKER_ID`、`SNOWFLAKE_DATACENTER_ID` | | WordPress | `WORDPRESS_BASE_URL`、`WORDPRESS_HOST_HEADER`、`WORDPRESS_ADMIN_USERNAME`、`WORDPRESS_ADMIN_PASSWORD`、`WORDPRESS_*_TIMEOUT_MS` | | Logging/Loki | `LOG_LEVEL`、`LOG_APP_NAME`、`LOKI_URL`、`LOKI_QUERY_HOST`、`LOKI_*` | -| QQBot/NapCat | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_*`、`QQBOT_SEND_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*`、`NAPCAT_*`、`QQBOT_NAPCAT_*`、`MQTT_*` | +| QQBot/NapCat | `QQBOT_ENABLED`、`QQBOT_ACCOUNT_SECRET_KEY`、`QQBOT_REVERSE_WS_*`、`QQBOT_SEND_*`、`QQBOT_PLUGIN_QUEUE_REDIS_*`、`QQBOT_COMMAND_MIN_COOLDOWN_MS`、`QQBOT_RULE_MIN_COOLDOWN_MS`、`QQBOT_REPEATER_*`、`NAPCAT_*`、`QQBOT_NAPCAT_*`、`MQTT_*` | | BangDream | `BANGDREAM_TSUGU_MAIN_SERVER`、`BANGDREAM_TSUGU_DISPLAYED_SERVERS`、`BANGDREAM_TSUGU_CACHE_ROOT` | | FF14 Market | `FF14_XIVAPI_BASE_URL`、`FF14_UNIVERSALIS_BASE_URL`、`FF14_MARKET_CACHE_TTL_MS` | | FFLogs | `FFLOGS_BASE_URL`、`FFLOGS_GRAPHQL_URL`、`FFLOGS_TOKEN_URL`、`FFLOGS_CLIENT_ID`、`FFLOGS_CLIENT_SECRET` | `DB_SYNC=true` 只适合本地开发或明确允许自动同步表结构的环境；生产应关闭并使用 SQL/迁移脚本。 +QQBot 插件 worker 使用 BullMQ 队列串行执行同一插件安装实例的请求。K8s 生产清单包含内部服务 `kt-qqbot-plugin-redis`，生产 env 可将 `QQBOT_PLUGIN_QUEUE_REDIS_HOST` 配为该服务名。 + ## 启动 ```bash diff --git a/k8s/prod/api.yaml b/k8s/prod/api.yaml index 2ab6b43..266b2b1 100644 --- a/k8s/prod/api.yaml +++ b/k8s/prod/api.yaml @@ -9,10 +9,7 @@ spec: replicas: 1 revisionHistoryLimit: 3 strategy: - type: RollingUpdate - rollingUpdate: - maxSurge: 1 - maxUnavailable: 0 + type: Recreate selector: matchLabels: app: kt-template-online-api @@ -87,3 +84,78 @@ spec: port: 48085 targetPort: 48085 nodePort: 30085 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: kt-qqbot-plugin-redis + namespace: kt-prod + labels: + app: kt-qqbot-plugin-redis +spec: + replicas: 1 + revisionHistoryLimit: 2 + selector: + matchLabels: + app: kt-qqbot-plugin-redis + template: + metadata: + labels: + app: kt-qqbot-plugin-redis + spec: + containers: + - name: redis + image: redis:7.4-alpine + imagePullPolicy: IfNotPresent + args: + - redis-server + - --appendonly + - "no" + - --save + - "" + ports: + - name: redis + containerPort: 6379 + readinessProbe: + tcpSocket: + port: 6379 + initialDelaySeconds: 3 + periodSeconds: 5 + timeoutSeconds: 2 + failureThreshold: 6 + livenessProbe: + tcpSocket: + port: 6379 + initialDelaySeconds: 15 + periodSeconds: 10 + timeoutSeconds: 2 + failureThreshold: 3 + resources: + requests: + cpu: 25m + memory: 64Mi + limits: + cpu: 250m + memory: 256Mi + volumeMounts: + - name: redis-data + mountPath: /data + volumes: + - name: redis-data + emptyDir: {} +--- +apiVersion: v1 +kind: Service +metadata: + name: kt-qqbot-plugin-redis + namespace: kt-prod + labels: + app: kt-qqbot-plugin-redis +spec: + type: ClusterIP + selector: + app: kt-qqbot-plugin-redis + ports: + - name: redis + port: 6379 + targetPort: 6379 diff --git a/package.json b/package.json index 2860d8c..92b3b76 100644 --- a/package.json +++ b/package.json @@ -34,6 +34,7 @@ }, "dependencies": { "@kwitsukasa/knife4j-swagger-vue3": "0.1.2", + "@nestjs/bullmq": "11.0.4", "@nestjs/common": "^11.1.24", "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.24", @@ -41,6 +42,7 @@ "@nestjs/swagger": "^11.4.4", "@nestjs/typeorm": "^11.0.1", "axios": "^1.17.0", + "bullmq": "5.78.1", "chart.js": "^4.5.1", "chartjs-adapter-moment": "^1.0.1", "cross-env": "^7.0.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 395f023..5c7b430 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -11,6 +11,9 @@ importers: '@kwitsukasa/knife4j-swagger-vue3': specifier: 0.1.2 version: 0.1.2(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2)) + '@nestjs/bullmq': + specifier: 11.0.4 + version: 11.0.4(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(bullmq@5.78.1) '@nestjs/common': specifier: ^11.1.24 version: 11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2) @@ -28,10 +31,13 @@ importers: version: 11.4.4(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(reflect-metadata@0.2.2) '@nestjs/typeorm': specifier: ^11.0.1 - version: 11.0.1(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2)(typeorm@0.3.30(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3))) + version: 11.0.1(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2)(typeorm@0.3.30(ioredis@5.10.1)(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3))) axios: specifier: ^1.17.0 version: 1.17.0 + bullmq: + specifier: 5.78.1 + version: 5.78.1 chart.js: specifier: ^4.5.1 version: 4.5.1 @@ -118,7 +124,7 @@ importers: version: 1.0.0-beta.2 typeorm: specifier: ^0.3.28 - version: 0.3.30(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)) + version: 0.3.30(ioredis@5.10.1)(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)) unified: specifier: ^11.0.5 version: 11.0.5 @@ -585,6 +591,9 @@ packages: '@types/node': optional: true + '@ioredis/commands@1.5.1': + resolution: {integrity: sha512-JH8ZL/ywcJyR9MmJ5BNqZllXNZQqQbnVZOqpPQqE1vHiFgAw4NHbvE0FOduNU8IX9babitBT46571OnPTT0Zcw==} + '@isaacs/cliui@8.0.2': resolution: {integrity: sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==} engines: {node: '>=12'} @@ -869,6 +878,49 @@ packages: '@microsoft/tsdoc@0.16.0': resolution: {integrity: sha512-xgAyonlVVS+q7Vc7qLW0UrJU7rSFcETRWsqdXZtjzRU8dF+6CkozTK4V4y1LwOX7j8r/vHphjDeMeGI4tNGeGA==} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + resolution: {integrity: sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==} + cpu: [arm64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + resolution: {integrity: sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==} + cpu: [x64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + resolution: {integrity: sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==} + cpu: [arm64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + resolution: {integrity: sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==} + cpu: [arm] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + resolution: {integrity: sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==} + cpu: [x64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + resolution: {integrity: sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==} + cpu: [x64] + os: [win32] + + '@nestjs/bull-shared@11.0.4': + resolution: {integrity: sha512-VBJcDHSAzxQnpcDfA0kt9MTGUD1XZzfByV70su0W0eDCQ9aqIEBlzWRW21tv9FG9dIut22ysgDidshdjlnczLw==} + peerDependencies: + '@nestjs/common': ^10.0.0 || ^11.0.0 + '@nestjs/core': ^10.0.0 || ^11.0.0 + + '@nestjs/bullmq@11.0.4': + resolution: {integrity: sha512-wBzK9raAVG0/6NTMdvLGM4/FQ1lsB35/pYS8L6a0SDgkTiLpd7mAjQ8R692oMx5s7IjvgntaZOuTUrKYLNfIkA==} + peerDependencies: + '@nestjs/common': ^10.0.0 || ^11.0.0 + '@nestjs/core': ^10.0.0 || ^11.0.0 + bullmq: ^3.0.0 || ^4.0.0 || ^5.0.0 + '@nestjs/cli@11.0.21': resolution: {integrity: sha512-F8mV0Sj/zVEouzR3NxBuJy08YHTUOmC5Xdcx3qIIaJWzrm8Vw86CHkhkaPBJ5ewRMHPDCShPmhsfwhpCcjts3A==} engines: {node: '>= 20.11'} @@ -1652,6 +1704,15 @@ packages: buffer@6.0.3: resolution: {integrity: sha512-FTiCpNxtwiZZHEZbcbTIcZjERVICn9yq/pDFkTl95/AxzD1naBctN7YO68riM/gLSDY7sdrMby8hofADYuuqOA==} + bullmq@5.78.1: + resolution: {integrity: sha512-zD5IT+qMqbMgPFPdL9FwnZka1bz6nckM+5lXj4N0vsXqdzoVO6wizmXpwsg/4GnHmXJsL7XOKeWA64tYUdPrOA==} + engines: {node: '>=12.22.0'} + peerDependencies: + redis: '>=5.0.0' + peerDependenciesMeta: + redis: + optional: true + busboy@1.6.0: resolution: {integrity: sha512-8SFQbg/0hQ9xy3UNTB0YEnsNBbWfhf7RtnzpL7TkBiTBRfrQ9Fxcnz7VJsleJpyp6rVLvXiuORqjlHi5q+PYuA==} engines: {node: '>=10.16.0'} @@ -1777,6 +1838,10 @@ packages: resolution: {integrity: sha512-JQHZ2QMW6l3aH/j6xCqQThY/9OH4D/9ls34cgkUBiEeocRTU04tHfKPBsUK1PqZCUQM7GiA0IIXJSuXHI64Kbg==} engines: {node: '>=0.8'} + cluster-key-slot@1.1.2: + resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==} + engines: {node: '>=0.10.0'} + co@4.6.0: resolution: {integrity: sha512-QVb0dM5HvG+uaxitm8wONl7jltx8dqhfU33DcqtOZcLSVIKSDDLDi7+0LbAKiyI8hD9u42m2YxXSkMGWThaecQ==} engines: {iojs: '>= 1.0.0', node: '>= 0.12.0'} @@ -1892,6 +1957,10 @@ packages: create-require@1.1.1: resolution: {integrity: sha512-dcKFX3jn0MpIaXjisoRvexIJVEKzaq7z2rZKxf+MSr9TkdmHmsU4m2lcLojrj/FHl8mk5VxMmYA+ftRkP/3oKQ==} + cron-parser@4.9.0: + resolution: {integrity: sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==} + engines: {node: '>=12.0.0'} + cross-env@7.0.3: resolution: {integrity: sha512-+/HKd6EgcQCJGh2PSjZuUitQBQynKor4wrFbRg4DtAgS1aWO+gU52xpH7M9ScGgXSYmAVS9bIJ8EzuaGw0oNAw==} engines: {node: '>=10.14', npm: '>=6', yarn: '>=1'} @@ -2575,6 +2644,10 @@ packages: inherits@2.0.4: resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==} + ioredis@5.10.1: + resolution: {integrity: sha512-HuEDBTI70aYdx1v6U97SbNx9F1+svQKBDo30o0b9fw055LMepzpOOd0Ccg9Q6tbqmBSJaMuY0fB7yw9/vjBYCA==} + engines: {node: '>=12.22.0'} + ip-address@10.2.0: resolution: {integrity: sha512-/+S6j4E9AHvW9SWMSEY9Xfy66O5PWvVEJ08O0y5JGyEKQpojb0K0GKpz/v5HJ/G0vi3D2sjGK78119oXZeE0qA==} engines: {node: '>= 12'} @@ -2932,6 +3005,12 @@ packages: resolution: {integrity: sha512-iPZK6eYjbxRu3uB4/WZ3EsEIMJFMqAoopl3R+zuq0UjcAm/MO6KCweDgPfP3elTztoKP3KtnVHxTn2NHBSDVUw==} engines: {node: '>=10'} + lodash.defaults@4.2.0: + resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==} + + lodash.isarguments@3.1.0: + resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==} + lodash.memoize@4.1.2: resolution: {integrity: sha512-t7j+NzmgnQzTAYXcsHYLgimltOV1MXHtlOWf6GjL9Kj8GK5FInw5JotxvbOs+IvV1/Dzo04/fCGfLVs7aXb4Ag==} @@ -2965,6 +3044,10 @@ packages: resolution: {integrity: sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA==} engines: {bun: '>=1.0.0', deno: '>=1.30.0', node: '>=8.0.0'} + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + magic-string@0.30.17: resolution: {integrity: sha512-sNPKHvyjVf7gyjwS4xGTaW/mCnF8wnjtifKBEhxfZ7E/S8tQ0rssrwGNn6q8JH/ohItJfSQp9mBtQYuTlH5QnA==} @@ -3234,6 +3317,13 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} + msgpackr-extract@3.0.4: + resolution: {integrity: sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==} + hasBin: true + + msgpackr@2.0.2: + resolution: {integrity: sha512-c5hYOXFbP79Slh6Dzd2wzk+jnV7mX1UxfMYtilnY1NmalXPqG8DGb5cYCMBrW4AsH3zekBBZd4QrKz9NhtvYLQ==} + multer@2.1.1: resolution: {integrity: sha512-mo+QTzKlx8R7E5ylSXxWzGoXoZbOsRMpyitcht8By2KHvMbf3tjwosZ/Mu/XYU6UuJ3VZnODIrak5ZrPiPyB6A==} engines: {node: '>= 10.16.0'} @@ -3292,6 +3382,10 @@ packages: encoding: optional: true + node-gyp-build-optional-packages@5.2.2: + resolution: {integrity: sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==} + hasBin: true + node-int64@0.4.0: resolution: {integrity: sha512-O5lz91xSOeoXP6DulyHfllpq+Eg00MWitZIbtPfoSEvqIHdl5gfcY6hYzDWnj0qD5tz52PI08u9qUvSVeUBeHw==} @@ -3625,6 +3719,14 @@ packages: real-require@1.0.0: resolution: {integrity: sha512-P4nbQYQfePJxRSmY+v/KINxVucm4NF3p3s7pJveMTtom52FR4YGltUQLB8idDXwDDWW+eYrWDFbuzUnjoWHF7g==} + redis-errors@1.2.0: + resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==} + engines: {node: '>=4'} + + redis-parser@3.0.0: + resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} + engines: {node: '>=4'} + reflect-metadata@0.1.14: resolution: {integrity: sha512-ZhYeb6nRaXCfhnndflDK8qI6ZQ/YcWZCISRAWICW9XYqMUwjZM9Z0DveWX/ABN01oxSHwVxKQmxeYZSsm0jh5A==} @@ -3762,6 +3864,11 @@ packages: resolution: {integrity: sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA==} hasBin: true + semver@7.8.1: + resolution: {integrity: sha512-rkVq3IXh+4FDGch+KwzX3aV9W3kO54GyEgpvBzSyctDA6Xtd7RJQV1xmXbeQp5v7+VzLOfVqiutSE6GICgPFvg==} + engines: {node: '>=10'} + hasBin: true + semver@7.8.2: resolution: {integrity: sha512-c8jsqUZm3omBOI66G90z1Dyw5z622G8oLG+omfsHBJf3CWQTlOcwOjvOG6wtiNfW6anKm/eA39LMwMtMez2TiQ==} engines: {node: '>=10'} @@ -3896,6 +4003,9 @@ packages: resolution: {integrity: sha512-XlkWvfIm6RmsWtNJx+uqtKLS8eqFbxUg0ZzLXqY0caEy9l7hruX8IpiDnjsLavoBgqCCR71TqWO8MaXYheJ3RQ==} engines: {node: '>=10'} + standard-as-callback@2.1.0: + resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==} + statuses@2.0.2: resolution: {integrity: sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw==} engines: {node: '>= 0.8'} @@ -4976,6 +5086,8 @@ snapshots: optionalDependencies: '@types/node': 22.19.19 + '@ioredis/commands@1.5.1': {} + '@isaacs/cliui@8.0.2': dependencies: string-width: 5.1.2 @@ -5460,6 +5572,38 @@ snapshots: '@microsoft/tsdoc@0.16.0': {} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + optional: true + + '@nestjs/bull-shared@11.0.4(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)': + dependencies: + '@nestjs/common': 11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 11.1.24(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2) + tslib: 2.8.1 + + '@nestjs/bullmq@11.0.4(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(bullmq@5.78.1)': + dependencies: + '@nestjs/bull-shared': 11.0.4(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24) + '@nestjs/common': 11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 11.1.24(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2) + bullmq: 5.78.1 + tslib: 2.8.1 + '@nestjs/cli@11.0.21(@types/node@22.19.19)(prettier@3.8.3)': dependencies: '@angular-devkit/core': 19.2.24(chokidar@4.0.3) @@ -5580,13 +5724,13 @@ snapshots: optionalDependencies: '@nestjs/platform-express': 11.1.24(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24) - '@nestjs/typeorm@11.0.1(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2)(typeorm@0.3.30(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)))': + '@nestjs/typeorm@11.0.1(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2)(typeorm@0.3.30(ioredis@5.10.1)(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)))': dependencies: '@nestjs/common': 11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2) '@nestjs/core': 11.1.24(@nestjs/common@11.1.24(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.1.24)(reflect-metadata@0.2.2)(rxjs@7.8.2) reflect-metadata: 0.2.2 rxjs: 7.8.2 - typeorm: 0.3.30(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)) + typeorm: 0.3.30(ioredis@5.10.1)(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)) '@noble/hashes@1.8.0': {} @@ -6330,6 +6474,17 @@ snapshots: base64-js: 1.5.1 ieee754: 1.2.1 + bullmq@5.78.1: + dependencies: + cron-parser: 4.9.0 + ioredis: 5.10.1 + msgpackr: 2.0.2 + node-abort-controller: 3.1.1 + semver: 7.8.1 + tslib: 2.8.1 + transitivePeerDependencies: + - supports-color + busboy@1.6.0: dependencies: streamsearch: 1.1.0 @@ -6440,6 +6595,8 @@ snapshots: clone@1.0.4: {} + cluster-key-slot@1.1.2: {} + co@4.6.0: {} codepage@1.15.0: {} @@ -6535,6 +6692,10 @@ snapshots: create-require@1.1.1: {} + cron-parser@4.9.0: + dependencies: + luxon: 3.7.2 + cross-env@7.0.3: dependencies: cross-spawn: 7.0.6 @@ -7331,6 +7492,20 @@ snapshots: inherits@2.0.4: {} + ioredis@5.10.1: + dependencies: + '@ioredis/commands': 1.5.1 + cluster-key-slot: 1.1.2 + debug: 4.4.3 + denque: 2.1.0 + lodash.defaults: 4.2.0 + lodash.isarguments: 3.1.0 + redis-errors: 1.2.0 + redis-parser: 3.0.0 + standard-as-callback: 2.1.0 + transitivePeerDependencies: + - supports-color + ip-address@10.2.0: {} ipaddr.js@1.9.1: {} @@ -7860,6 +8035,10 @@ snapshots: dependencies: p-locate: 5.0.0 + lodash.defaults@4.2.0: {} + + lodash.isarguments@3.1.0: {} + lodash.memoize@4.1.2: {} lodash.merge@4.6.2: {} @@ -7885,6 +8064,8 @@ snapshots: lru.min@1.1.4: {} + luxon@3.7.2: {} + magic-string@0.30.17: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -8344,6 +8525,22 @@ snapshots: ms@2.1.3: {} + msgpackr-extract@3.0.4: + dependencies: + node-gyp-build-optional-packages: 5.2.2 + optionalDependencies: + '@msgpackr-extract/msgpackr-extract-darwin-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-darwin-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-win32-x64': 3.0.4 + optional: true + + msgpackr@2.0.2: + optionalDependencies: + msgpackr-extract: 3.0.4 + multer@2.1.1: dependencies: append-field: 1.0.0 @@ -8400,6 +8597,11 @@ snapshots: dependencies: whatwg-url: 5.0.0 + node-gyp-build-optional-packages@5.2.2: + dependencies: + detect-libc: 2.1.2 + optional: true + node-int64@0.4.0: {} node-releases@2.0.47: {} @@ -8730,6 +8932,12 @@ snapshots: real-require@1.0.0: {} + redis-errors@1.2.0: {} + + redis-parser@3.0.0: + dependencies: + redis-errors: 1.2.0 + reflect-metadata@0.1.14: {} reflect-metadata@0.2.2: {} @@ -8897,6 +9105,8 @@ snapshots: semver@6.3.1: {} + semver@7.8.1: {} + semver@7.8.2: {} send@1.2.1: @@ -9052,6 +9262,8 @@ snapshots: dependencies: escape-string-regexp: 2.0.0 + standard-as-callback@2.1.0: {} + statuses@2.0.2: {} streamsearch@1.1.0: {} @@ -9362,7 +9574,7 @@ snapshots: typedarray@0.0.6: {} - typeorm@0.3.30(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)): + typeorm@0.3.30(ioredis@5.10.1)(mysql2@3.22.4(@types/node@22.19.19))(ts-node@10.9.2(@types/node@22.19.19)(typescript@5.9.3)): dependencies: '@sqltools/formatter': 1.2.5 ansis: 4.3.1 @@ -9380,6 +9592,7 @@ snapshots: uuid: 11.1.1 yargs: 17.7.2 optionalDependencies: + ioredis: 5.10.1 mysql2: 3.22.4(@types/node@22.19.19) ts-node: 10.9.2(@types/node@22.19.19)(typescript@5.9.3) transitivePeerDependencies: diff --git a/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts b/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts index fb0c34b..7e8a5bc 100644 --- a/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts +++ b/src/modules/qqbot/plugin-platform/application/plugin-platform.service.ts @@ -16,7 +16,10 @@ import { parseQqbotPluginManifest, type QqbotPluginManifest, } from '../domain/manifest'; -import type { QqbotPluginWorkerRuntime } from '../infrastructure/integration/runtime'; +import type { + QqbotPluginRuntimeEvent as QqbotPluginWorkerRuntimeEvent, + QqbotPluginWorkerRuntime, +} from '../infrastructure/integration/runtime'; import { QqbotPluginArgumentParserService } from './argument/qqbot-plugin-argument-parser.service'; import { QqbotBuiltinPluginPackageLoaderService } from '../infrastructure/integration/package/builtin-plugin-package-loader.service'; import { QqbotPluginPackageReaderService } from '../infrastructure/integration/package/plugin-package-reader.service'; @@ -51,6 +54,7 @@ export type QqbotPluginRuntimeFactory = { | 'activate' | 'deactivate' | 'dispose' + | 'drainRuntimeEvents' | 'executeOperation' | 'handleEvent' | 'health' @@ -351,30 +355,34 @@ export class QqbotPluginPlatformService ); } - const output = await workerContext.worker.executeOperation({ - input: normalizedInput, - operationId: operation.key, - operationKey: operation.key, - timeoutMs: operation.timeoutMs, - }); - const pluginId = this.getPluginIdFromContext(input.context); - if (pluginId) { - await this.runtimeEventRepository.save({ - eventType: 'command log mapped', - installationId: null, - level: 'info', - pluginId, - safeSummary: { - operationKey: input.operationKey, - outputKeys: - output && typeof output === 'object' - ? Object.keys(output as Record).sort() - : [], - pluginKey: input.pluginKey, - }, + try { + const output = await workerContext.worker.executeOperation({ + input: normalizedInput, + operationId: operation.key, + operationKey: operation.key, + timeoutMs: operation.timeoutMs, }); + const pluginId = this.getPluginIdFromContext(input.context); + if (pluginId) { + await this.runtimeEventRepository.save({ + eventType: 'command log mapped', + installationId: null, + level: 'info', + pluginId, + safeSummary: { + operationKey: input.operationKey, + outputKeys: + output && typeof output === 'object' + ? Object.keys(output as Record).sort() + : [], + pluginKey: input.pluginKey, + }, + }); + } + return output; + } finally { + await this.flushWorkerRuntimeEvents(workerContext); } - return output; } async dispatchEvent(input: QqbotPluginEventDispatchInput) { @@ -387,12 +395,16 @@ export class QqbotPluginPlatformService ) { continue; } - const result = await workerContext.worker.handleEvent({ - event: input.message, - eventKey: event.eventName || input.eventKey, - timeoutMs: workerContext.manifest.runtime.timeoutMs, - }); - handled = Boolean(result) || handled; + try { + const result = await workerContext.worker.handleEvent({ + event: input.message, + eventKey: event.eventName || input.eventKey, + timeoutMs: workerContext.manifest.runtime.timeoutMs, + }); + handled = Boolean(result) || handled; + } finally { + await this.flushWorkerRuntimeEvents(workerContext); + } } } return handled; @@ -745,11 +757,20 @@ export class QqbotPluginPlatformService private async stopWorker(installationId: string) { const worker = this.activeWorkers.get(installationId); if (!worker) return; + const workerContext = this.activeWorkerContexts.get(installationId); try { - await worker.deactivate(); + try { + await worker.deactivate(); + } finally { + await this.flushWorkerRuntimeEventsBestEffort(workerContext); + } } finally { - await worker.dispose(); - this.unregisterActiveWorker(installationId); + try { + await worker.dispose(); + } finally { + await this.flushWorkerRuntimeEventsBestEffort(workerContext); + this.unregisterActiveWorker(installationId); + } } } @@ -788,17 +809,79 @@ export class QqbotPluginPlatformService await worker.load(version.manifestJson); await worker.activate(); await worker.health(); + await this.flushRuntimeEvents( + installation.id, + installation.pluginId, + worker, + ); await this.registerActiveWorker( installation, version, worker as QqbotPluginWorkerRuntime, ); } catch (error) { - await worker.dispose(); + await this.flushRuntimeEvents( + installation.id, + installation.pluginId, + worker, + ); + try { + await worker.dispose(); + } finally { + await this.flushRuntimeEvents( + installation.id, + installation.pluginId, + worker, + ); + } throw error; } } + private async flushWorkerRuntimeEvents(workerContext: ActiveWorkerContext) { + await this.flushRuntimeEvents( + workerContext.installationId, + workerContext.pluginId, + workerContext.worker, + ); + } + + private async flushWorkerRuntimeEventsBestEffort( + workerContext?: ActiveWorkerContext, + ) { + if (!workerContext) return; + try { + await this.flushWorkerRuntimeEvents(workerContext); + } catch { + // Runtime event persistence must not block worker cleanup. + } + } + + private async flushRuntimeEvents( + installationId: string, + pluginId: string, + worker: Pick, + ) { + const events = worker.drainRuntimeEvents?.() || []; + if (!events.length || !this.isPersistablePluginId(pluginId)) return; + + await Promise.all( + events.map((event: QqbotPluginWorkerRuntimeEvent) => + this.runtimeEventRepository.save({ + eventType: event.eventType, + installationId, + level: event.level, + pluginId, + safeSummary: event.safeSummary, + }), + ), + ); + } + + private isPersistablePluginId(pluginId: string) { + return /^\d+$/.test(pluginId); + } + private async recordRuntimeEvent( installation: QqbotPluginInstallation, eventType: string, diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker-runtime.factory.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker-runtime.factory.ts index a316490..4d6f829 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker-runtime.factory.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker-runtime.factory.ts @@ -1,4 +1,5 @@ import { Injectable } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; import { Worker } from 'node:worker_threads'; import { join } from 'node:path'; import { throwVbenError } from '@/common'; @@ -13,8 +14,14 @@ import type { QqbotPluginVersion, } from '@/modules/qqbot/plugin-platform/infrastructure/persistence'; import { + QqbotPluginWorkerResponseError, QqbotPluginWorkerRuntime, + serializePluginWorkerResponseError, } from './worker-runtime'; +import { + createQqbotBullmqWorkerQueueOptions, + QqbotBullmqPluginWorkerRequestQueue, +} from './bullmq-plugin-worker-request.queue'; import type { QqbotPluginWorkerDriver, QqbotPluginWorkerRequest, @@ -26,6 +33,7 @@ export class QqbotBuiltinPluginWorkerRuntimeFactoryService { constructor( private readonly pluginLoader: QqbotBuiltinPluginPackageLoaderService, + private readonly configService: ConfigService, ) {} create( @@ -33,8 +41,19 @@ export class QqbotBuiltinPluginWorkerRuntimeFactoryService version: QqbotPluginVersion, ) { const pluginKey = getManifestPluginKey(version.manifestJson); + const driver = new QqbotBuiltinPluginWorkerThreadDriver( + this.pluginLoader, + pluginKey, + ); return new QqbotPluginWorkerRuntime( - new QqbotBuiltinPluginWorkerThreadDriver(this.pluginLoader, pluginKey), + new QqbotBullmqPluginWorkerRequestQueue( + driver, + createQqbotBullmqWorkerQueueOptions( + this.configService, + pluginKey, + installation.id, + ), + ), { defaultTimeoutMs: getDefaultRuntimeTimeout(version.manifestJson), installationId: installation.id, @@ -133,7 +152,7 @@ export class QqbotBuiltinPluginWorkerThreadDriver pending.resolve(message.result); return; } - pending.reject(deserializeWorkerError(message.error)); + pending.reject(new QqbotPluginWorkerResponseError(message.error || {})); return; } @@ -177,20 +196,7 @@ function resolveWorkerExecArgv() { } function serializeWorkerError(error: unknown) { - return { - message: error instanceof Error ? error.message : `${error}`, - name: error instanceof Error ? error.name : 'Error', - stack: error instanceof Error ? error.stack : undefined, - }; -} - -function deserializeWorkerError( - error?: { message?: string; name?: string; stack?: string }, -) { - const output = new Error(error?.message || 'QQBot 插件 worker 请求失败'); - if (error?.name) output.name = error.name; - if (error?.stack) output.stack = error.stack; - return output; + return serializePluginWorkerResponseError(error); } function getManifestPluginKey(manifest: unknown) { diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts new file mode 100644 index 0000000..56cec68 --- /dev/null +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts @@ -0,0 +1,318 @@ +import { Logger } from '@nestjs/common'; +import type { ConfigService } from '@nestjs/config'; +import { + Job, + Queue, + QueueEvents, + Worker, + type ConnectionOptions, + type JobsOptions, +} from 'bullmq'; +import { + QqbotPluginWorkerExpiredRequestError, + QqbotPluginWorkerResponseError, + QqbotPluginWorkerStaleRequestError, + type QqbotPluginWorkerResponseErrorInput, +} from './worker-runtime'; +import type { + QqbotPluginWorkerDriver, + QqbotPluginWorkerRequest, + QqbotPluginWorkerRequestQueue, +} from './worker-runtime.types'; + +export type QqbotBullmqWorkerRequestQueueOptions = { + connection: ConnectionOptions; + installationId: string; + pluginKey: string; + prefix: string; + removeOnFailCount: number; + waitUntilFinishedBufferMs: number; + workerInstanceId: string; +}; + +type QqbotWorkerQueueJobData = { + expiresAt: number; + generation: number; + message: QqbotPluginWorkerRequest; + workerInstanceId: string; +}; + +type QqbotWorkerQueueResult = + | { + ok: true; + result: unknown; + } + | { + error: QqbotPluginWorkerResponseErrorInput; + ok: false; + }; + +export class QqbotBullmqPluginWorkerRequestQueue + implements QqbotPluginWorkerRequestQueue +{ + private readonly queue: Queue< + QqbotWorkerQueueJobData, + QqbotWorkerQueueResult, + string + >; + private readonly queueEvents: QueueEvents; + private readonly ready: Promise; + private readonly worker: Worker< + QqbotWorkerQueueJobData, + QqbotWorkerQueueResult, + string + >; + private readonly logger = new Logger( + QqbotBullmqPluginWorkerRequestQueue.name, + ); + private closed = false; + private generation = 0; + + constructor( + private readonly driver: QqbotPluginWorkerDriver, + private readonly options: QqbotBullmqWorkerRequestQueueOptions, + ) { + const queueName = buildWorkerQueueName( + options.pluginKey, + options.installationId, + ); + const bullmqOptions = { + connection: options.connection, + prefix: options.prefix, + }; + this.queue = new Queue(queueName, bullmqOptions); + this.queueEvents = new QueueEvents(queueName, bullmqOptions); + this.worker = new Worker( + queueName, + async (job: Job) => { + if (job.data.expiresAt <= Date.now()) { + return { + error: { + message: 'worker-request-expired', + name: 'QqbotPluginWorkerExpiredRequestError', + }, + ok: false, + }; + } + if (job.data.workerInstanceId !== this.options.workerInstanceId) { + return { + error: { + message: 'worker-request-foreign-instance', + name: 'QqbotPluginWorkerStaleRequestError', + }, + ok: false, + }; + } + if (job.data.generation !== this.generation) { + return { + error: { + message: 'worker-request-stale', + name: 'QqbotPluginWorkerStaleRequestError', + }, + ok: false, + }; + } + try { + return { + ok: true, + result: await this.driver.request(job.data.message), + }; + } catch (error) { + if (error instanceof QqbotPluginWorkerResponseError) { + return { + error: error.serializedError, + ok: false, + }; + } + await this.reset(); + throw error; + } + }, + { + ...bullmqOptions, + concurrency: 1, + }, + ); + this.queue.on('error', (error) => { + this.logBullmqError('queue', error); + }); + this.queueEvents.on('error', (error) => { + this.logBullmqError('queueEvents', error); + }); + this.worker.on('error', (error) => { + this.logBullmqError('worker', error); + }); + this.ready = Promise.all([ + this.queue.waitUntilReady(), + this.queueEvents.waitUntilReady(), + this.worker.waitUntilReady(), + ]); + } + + async request(message: QqbotPluginWorkerRequest): Promise { + if (this.closed) { + throw new Error('QQBot 插件 worker 队列已关闭'); + } + + await this.ready; + const job = await this.queue.add( + message.type, + { + expiresAt: + Date.now() + message.timeoutMs + this.options.waitUntilFinishedBufferMs, + generation: this.generation, + message, + workerInstanceId: this.options.workerInstanceId, + }, + { + attempts: 1, + jobId: message.correlationId, + removeOnComplete: true, + removeOnFail: this.options.removeOnFailCount, + } satisfies JobsOptions, + ); + const result = await job.waitUntilFinished( + this.queueEvents, + message.timeoutMs + this.options.waitUntilFinishedBufferMs, + ); + + if (result?.ok === false) { + if (result.error.name === 'QqbotPluginWorkerStaleRequestError') { + throw new QqbotPluginWorkerStaleRequestError(result.error.message); + } + if (result.error.name === 'QqbotPluginWorkerExpiredRequestError') { + throw new QqbotPluginWorkerExpiredRequestError(result.error.message); + } + throw new QqbotPluginWorkerResponseError(result.error); + } + return result?.result; + } + + async reset(): Promise { + this.generation += 1; + await this.driver.dispose(); + } + + async close(): Promise { + this.closed = true; + await Promise.allSettled([ + this.worker.close(), + this.queueEvents.close(), + this.queue.close(), + this.driver.dispose(), + ]); + } + + private logBullmqError( + source: 'queue' | 'queueEvents' | 'worker', + error: Error, + ) { + this.logger.error( + `QQBot 插件 worker BullMQ ${source} 异常：${error.message}`, + error.stack, + ); + } +} + +export function createQqbotBullmqWorkerQueueOptions( + configService: ConfigService, + pluginKey: string, + installationId: string, +): QqbotBullmqWorkerRequestQueueOptions { + return { + connection: resolveQqbotPluginQueueConnection(configService), + installationId, + pluginKey, + prefix: resolveQqbotPluginQueuePrefix(configService), + removeOnFailCount: readNumberConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_REMOVE_ON_FAIL'], + 100, + ), + waitUntilFinishedBufferMs: readNumberConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_WAIT_BUFFER_MS'], + 5_000, + ), + workerInstanceId: createWorkerInstanceId(), + }; +} + +export function resolveQqbotPluginQueuePrefix(configService: ConfigService) { + return readStringConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_REDIS_PREFIX'], + 'kt:qqbot:plugin-worker', + ); +} + +export function resolveQqbotPluginQueueConnection( + configService: ConfigService, +): ConnectionOptions { + const host = readStringConfig(configService, [ + 'QQBOT_PLUGIN_QUEUE_REDIS_HOST', + 'REDIS_HOST', + ]); + if (!host) { + throw new Error('QQBot 插件队列缺少 Redis 主机配置'); + } + + const password = readStringConfig(configService, [ + 'QQBOT_PLUGIN_QUEUE_REDIS_PASSWORD', + 'REDIS_PASSWORD', + ]); + return { + db: readNumberConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_REDIS_DB', 'REDIS_DB'], + 0, + ), + host, + password: password || undefined, + port: readNumberConfig( + configService, + ['QQBOT_PLUGIN_QUEUE_REDIS_PORT', 'REDIS_PORT'], + 6379, + ), + }; +} + +function buildWorkerQueueName(pluginKey: string, installationId: string) { + const safePluginKey = pluginKey.replace(/[^a-zA-Z0-9_-]/g, '-'); + const safeInstallationId = installationId.replace(/[^a-zA-Z0-9_-]/g, '-'); + return `qqbot-plugin-worker:${safePluginKey}:${safeInstallationId}`; +} + +function createWorkerInstanceId() { + return [ + process.env.HOSTNAME || 'local', + process.pid, + Date.now(), + Math.random().toString(16).slice(2), + ].join(':'); +} + +function readStringConfig( + configService: ConfigService, + keys: string[], + fallback = '', +) { + for (const key of keys) { + const value = configService.get(key); + if (value !== undefined && value !== null && `${value}`.trim()) { + return `${value}`.trim(); + } + } + return fallback; +} + +function readNumberConfig( + configService: ConfigService, + keys: string[], + fallback: number, +) { + const value = readStringConfig(configService, keys); + if (!value) return fallback; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : fallback; +} diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/index.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/index.ts index e3c7ec8..5ea93c2 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/index.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/index.ts @@ -1,3 +1,4 @@ export * from './worker-runtime'; export * from './worker-runtime.types'; export * from './builtin-plugin-worker-runtime.factory'; +export * from './bullmq-plugin-worker-request.queue'; diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts index 4b092f0..7186a61 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.ts @@ -5,7 +5,7 @@ import { type QqbotPluginRuntimeEvent, type QqbotPluginRuntimeStatus, type QqbotPluginSafeInputSummary, - type QqbotPluginWorkerDriver, + type QqbotPluginWorkerRequestQueue, type QqbotPluginWorkerRequest, type QqbotPluginWorkerRequestType, type QqbotPluginWorkerRuntimeOptions, @@ -23,6 +23,46 @@ export class QqbotPluginRuntimeError extends Error { } } +export type QqbotPluginWorkerResponseErrorInput = { + message?: string; + name?: string; + stack?: string; +}; + +export class QqbotPluginWorkerResponseError extends Error { + constructor(readonly serializedError: QqbotPluginWorkerResponseErrorInput) { + super(serializedError.message || 'QQBot 插件 worker 请求失败'); + this.name = serializedError.name || 'QqbotPluginWorkerResponseError'; + if (serializedError.stack) this.stack = serializedError.stack; + } +} + +export const serializePluginWorkerResponseError = ( + error: unknown, +): QqbotPluginWorkerResponseErrorInput => ({ + message: error instanceof Error ? error.message : `${error}`, + name: error instanceof Error ? error.name : 'Error', + stack: error instanceof Error ? error.stack : undefined, +}); + +export class QqbotPluginWorkerStaleRequestError extends Error { + constructor(message = 'QQBot 插件 worker 队列请求已过期，需要恢复后重试') { + super(message); + this.name = 'QqbotPluginWorkerStaleRequestError'; + } +} + +export class QqbotPluginWorkerExpiredRequestError extends Error { + constructor(message = 'QQBot 插件 worker 队列请求已超时') { + super(message); + this.name = 'QqbotPluginWorkerExpiredRequestError'; + } +} + +const isNamedError = (error: unknown, name: string) => { + return error instanceof Error && error.name === name; +}; + const createCorrelationId = () => { return `qqbot-plugin-${Date.now()}-${Math.random().toString(16).slice(2)}`; }; @@ -38,18 +78,27 @@ const summarizeInput = ( export class QqbotPluginWorkerRuntime { private readonly runtimeEvents: QqbotPluginRuntimeEvent[] = []; + private manifestForRecovery?: unknown; + private recoveryPromise?: Promise; + private shouldRecoverActive = false; status: QqbotPluginRuntimeStatus = 'stopped'; constructor( - private readonly driver: QqbotPluginWorkerDriver, + private readonly requestQueue: QqbotPluginWorkerRequestQueue, private readonly options: QqbotPluginWorkerRuntimeOptions, ) {} async load(manifest: unknown) { - const result = await this.request('load', { - manifest, - }); + this.manifestForRecovery = manifest; + const result = await this.request( + 'load', + { + manifest, + }, + undefined, + { skipRecovery: true }, + ); this.status = 'loaded'; return result; } @@ -57,6 +106,7 @@ export class QqbotPluginWorkerRuntime { async activate() { const result = await this.request('activate'); this.status = 'active'; + this.shouldRecoverActive = true; return result; } @@ -92,15 +142,17 @@ export class QqbotPluginWorkerRuntime { async deactivate() { const result = await this.request('deactivate'); this.status = 'stopped'; + this.shouldRecoverActive = false; return result; } async dispose() { try { - await this.request('dispose'); + await this.request('dispose', {}, undefined, { skipRecovery: true }); } finally { - await this.driver.dispose(); + await this.requestQueue.close(); this.status = 'stopped'; + this.shouldRecoverActive = false; } } @@ -108,11 +160,20 @@ export class QqbotPluginWorkerRuntime { return [...this.runtimeEvents]; } + drainRuntimeEvents() { + return this.runtimeEvents.splice(0); + } + private async request( type: QqbotPluginWorkerRequestType, payload: Partial = {}, timeoutMs = this.options.defaultTimeoutMs, + control: { retryStale?: boolean; skipRecovery?: boolean } = {}, ) { + if (!control.skipRecovery) { + await this.recoverIfNeeded(type); + } + const message: QqbotPluginWorkerRequest = { correlationId: createCorrelationId(), installationId: this.options.installationId, @@ -123,7 +184,7 @@ export class QqbotPluginWorkerRuntime { }; const requestPromise = Promise.resolve().then(() => - this.driver.request(message), + this.requestQueue.request(message), ); requestPromise.catch(() => undefined); @@ -135,8 +196,47 @@ export class QqbotPluginWorkerRuntime { if (error instanceof QqbotPluginRuntimeError) { throw error; } + if ( + (error instanceof QqbotPluginWorkerStaleRequestError || + isNamedError(error, 'QqbotPluginWorkerStaleRequestError')) && + control.retryStale !== false + ) { + await this.markWorkerFailed('worker-stale-request', { + correlationId: message.correlationId, + operationId: message.operationId, + type, + }); + await this.recoverIfNeeded(type); + return this.request(type, payload, timeoutMs, { + retryStale: false, + skipRecovery: true, + }); + } + if ( + error instanceof QqbotPluginWorkerExpiredRequestError || + isNamedError(error, 'QqbotPluginWorkerExpiredRequestError') + ) { + const runtimeError = new QqbotPluginRuntimeError( + 'PLUGIN_WORKER_TIMEOUT', + this.options.pluginKey, + 'QQBot plugin worker queue request expired.', + { + correlationId: message.correlationId, + operationId: message.operationId, + timeoutMs, + type, + }, + ); + this.recordRuntimeEvent( + 'worker-request-expired', + runtimeError.safeSummary, + ); + throw runtimeError; + } + if (error instanceof QqbotPluginWorkerResponseError) { + throw error; + } - this.status = 'failed'; const runtimeError = new QqbotPluginRuntimeError( 'PLUGIN_WORKER_CRASH', this.options.pluginKey, @@ -148,7 +248,7 @@ export class QqbotPluginWorkerRuntime { type, }, ); - this.recordRuntimeEvent('worker-crash', runtimeError.safeSummary); + await this.markWorkerFailed('worker-crash', runtimeError.safeSummary); throw runtimeError; } finally { timeout.clear(); @@ -163,7 +263,6 @@ export class QqbotPluginWorkerRuntime { let timer: NodeJS.Timeout | undefined; const promise = new Promise((_, reject) => { timer = setTimeout(async () => { - this.status = 'failed'; const error = new QqbotPluginRuntimeError( 'PLUGIN_WORKER_TIMEOUT', this.options.pluginKey, @@ -175,13 +274,7 @@ export class QqbotPluginWorkerRuntime { type, }, ); - this.recordRuntimeEvent('worker-timeout', error.safeSummary); - - try { - await this.driver.dispose(); - } catch { - // Timeout disposal is best-effort; the timeout error is the primary signal. - } + await this.markWorkerFailed('worker-timeout', error.safeSummary); reject(error); }, timeoutMs); timer.unref?.(); @@ -194,13 +287,87 @@ export class QqbotPluginWorkerRuntime { }; } - private recordRuntimeEvent( + private async recoverIfNeeded(triggerType: QqbotPluginWorkerRequestType) { + if (this.status !== 'failed' || !this.manifestForRecovery) return; + + if (!this.recoveryPromise) { + this.recoveryPromise = this.recoverWorker(triggerType).finally(() => { + this.recoveryPromise = undefined; + }); + } + + await this.recoveryPromise; + } + + private async recoverWorker(triggerType: QqbotPluginWorkerRequestType) { + this.recordRuntimeEvent( + 'worker-recover-started', + { + triggerType, + }, + 'warn', + ); + + await this.request( + 'load', + { + manifest: this.manifestForRecovery, + }, + this.options.defaultTimeoutMs, + { skipRecovery: true }, + ); + this.status = 'loaded'; + + if (this.shouldRecoverActive) { + await this.request( + 'activate', + {}, + this.options.defaultTimeoutMs, + { skipRecovery: true }, + ); + this.status = 'active'; + } + + this.recordRuntimeEvent( + 'worker-recovered', + { + status: this.status, + triggerType, + }, + 'info', + ); + } + + private async markWorkerFailed( eventType: string, safeSummary: Record, + ) { + if (this.status === 'active') { + this.shouldRecoverActive = true; + } + this.status = 'failed'; + + let resetError: string | undefined; + try { + await this.requestQueue.reset(); + } catch (error) { + resetError = error instanceof Error ? error.message : `${error}`; + } + + this.recordRuntimeEvent(eventType, { + ...safeSummary, + ...(resetError ? { resetError } : {}), + }); + } + + private recordRuntimeEvent( + eventType: string, + safeSummary: Record, + level: QqbotPluginRuntimeEvent['level'] = 'error', ) { this.runtimeEvents.push({ eventType, - level: 'error', + level, pluginKey: this.options.pluginKey, safeSummary, }); diff --git a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts index aed03a5..ba21ad4 100644 --- a/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts +++ b/src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/worker-runtime.types.ts @@ -38,6 +38,12 @@ export type QqbotPluginWorkerDriver = { request(message: QqbotPluginWorkerRequest): Promise; }; +export type QqbotPluginWorkerRequestQueue = { + close(): Promise; + request(message: QqbotPluginWorkerRequest): Promise; + reset(): Promise; +}; + export type QqbotPluginWorkerRuntimeOptions = { defaultTimeoutMs: number; installationId: string; diff --git a/src/modules/qqbot/plugin-platform/plugin-platform.module.ts b/src/modules/qqbot/plugin-platform/plugin-platform.module.ts index d313999..f9fb994 100644 --- a/src/modules/qqbot/plugin-platform/plugin-platform.module.ts +++ b/src/modules/qqbot/plugin-platform/plugin-platform.module.ts @@ -1,5 +1,6 @@ import { forwardRef, Module } from '@nestjs/common'; -import { ConfigModule } from '@nestjs/config'; +import { BullModule } from '@nestjs/bullmq'; +import { ConfigModule, ConfigService } from '@nestjs/config'; import { TypeOrmModule } from '@nestjs/typeorm'; import { AdminAuthGuardModule } from '@/modules/admin/identity/auth/admin-auth-guard.module'; import { DictModule } from '@/modules/admin/platform-config/dict/dict.module'; @@ -16,7 +17,11 @@ import { QQBOT_PLUGIN_PLATFORM_ENTITIES } from './infrastructure/persistence'; import { QqbotBuiltinPluginPackageLoaderService } from './infrastructure/integration/package/builtin-plugin-package-loader.service'; import { QqbotPluginPackageReaderService } from './infrastructure/integration/package/plugin-package-reader.service'; import { QqbotPluginHttpClientService } from './infrastructure/integration/sdk'; -import { QqbotBuiltinPluginWorkerRuntimeFactoryService } from './infrastructure/integration/runtime'; +import { + QqbotBuiltinPluginWorkerRuntimeFactoryService, + resolveQqbotPluginQueueConnection, + resolveQqbotPluginQueuePrefix, +} from './infrastructure/integration/runtime'; import { QQBOT_PLUGIN_RUNTIME_FACTORY } from './application/plugin-platform.service'; @Module({ @@ -28,6 +33,14 @@ import { QQBOT_PLUGIN_RUNTIME_FACTORY } from './application/plugin-platform.serv ], imports: [ ConfigModule, + BullModule.forRootAsync({ + imports: [ConfigModule], + inject: [ConfigService], + useFactory: (configService: ConfigService) => ({ + connection: resolveQqbotPluginQueueConnection(configService), + prefix: resolveQqbotPluginQueuePrefix(configService), + }), + }), AdminAuthGuardModule, DictModule, forwardRef(() => QqbotCoreModule), diff --git a/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts b/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts index d881638..8fddfda 100644 --- a/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts +++ b/test/modules/qqbot/plugin-platform/plugin-lifecycle-runtime.spec.ts @@ -1,6 +1,7 @@ import { existsSync, readFileSync, readdirSync, statSync } from 'fs'; import { join } from 'path'; import { MODULE_METADATA } from '@nestjs/common/constants'; +import { ConfigService } from '@nestjs/config'; import { QQBOT_PLUGIN_RUNTIME_FACTORY } from '../../../../src/modules/qqbot/plugin-platform/application/plugin-platform.service'; import { QqbotPluginPlatformService } from '../../../../src/modules/qqbot/plugin-platform/application/plugin-platform.service'; import { QqbotEventPluginRegistryService } from '../../../../src/modules/qqbot/plugin-platform/application/registry/qqbot-event-plugin-registry.service'; @@ -49,6 +50,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { ); expect(dependencies?.[0]).toBe(QqbotBuiltinPluginPackageLoaderService); + expect(dependencies?.[1]).toBe(ConfigService); }); it('uses a real worker-thread boundary for built-in plugin runtimes', () => { @@ -62,6 +64,45 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { expect(source).not.toContain('new QqbotBuiltinPluginWorkerDriver'); }); + it('uses BullMQ queues to serialize plugin worker requests instead of ad hoc in-memory chaining', () => { + const source = [ + readSource( + 'src/modules/qqbot/plugin-platform/plugin-platform.module.ts', + ), + readSource( + 'src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/bullmq-plugin-worker-request.queue.ts', + ), + readSource( + 'src/modules/qqbot/plugin-platform/infrastructure/integration/runtime/builtin-plugin-worker-runtime.factory.ts', + ), + ].join('\n'); + + expect(source).toContain("@nestjs/bullmq"); + expect(source).toContain("from 'bullmq'"); + expect(source).toContain('new Queue('); + expect(source).toContain('new Worker('); + expect(source).toContain('new QueueEvents('); + expect(source).toContain('concurrency: 1'); + expect(source).toContain('installation.id'); + expect(source).toContain('options.installationId'); + expect(source).toContain("this.queue.on('error'"); + expect(source).toContain("this.queueEvents.on('error'"); + expect(source).toContain("this.worker.on('error'"); + expect(source).toContain('expiresAt'); + expect(source).toContain('workerInstanceId'); + expect(source).toContain('worker-request-expired'); + expect(source).not.toContain('previous.catch(() => undefined).then'); + }); + + it('keeps API deployment as a single plugin queue consumer during releases', () => { + const source = readSource('k8s/prod/api.yaml'); + + expect(source).toContain('replicas: 1'); + expect(source).toContain('type: Recreate'); + expect(source).not.toContain('maxSurge: 1'); + expect(source).not.toContain('maxUnavailable: 0'); + }); + it('uses dedicated lifecycle use cases instead of direct status flips', () => { const controller = readSource( 'src/modules/qqbot/plugin-platform/contract/plugin-platform.controller.ts', @@ -181,7 +222,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { const installation = { id: 'installation-1', installedPath: 'D:/plugins/demo-plugin', - pluginId: 'plugin-1', + pluginId: '2060000000000000002', runtimeStatus: 'stopped', status: 'installed', versionId: 'version-1', @@ -212,10 +253,24 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { const configRepository = createRepository(); const assetRepository = createRepository(); const runtimeEventRepository = createRepository(); + const runtimeEventBatches = [ + [], + [ + { + eventType: 'worker-dispose-finished', + level: 'info', + pluginKey: 'demo-plugin', + safeSummary: { + phase: 'dispose', + }, + }, + ], + ]; const worker = { activate: jest.fn(async () => ({ ok: true })), deactivate: jest.fn(async () => ({ ok: true })), dispose: jest.fn(async () => undefined), + drainRuntimeEvents: jest.fn(() => runtimeEventBatches.shift() || []), health: jest.fn(async () => ({ ok: true })), load: jest.fn(async () => ({ ok: true })), }; @@ -290,6 +345,132 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { ); expect(worker.deactivate).toHaveBeenCalled(); expect(worker.dispose).toHaveBeenCalled(); + expect(runtimeEventRepository.save).toHaveBeenCalledWith( + expect.objectContaining({ + eventType: 'worker-dispose-finished', + installationId: installation.id, + pluginId: installation.pluginId, + }), + ); + }); + + it('keeps worker cleanup best-effort when stop runtime event persistence fails', async () => { + const manifest = { + assets: [], + configSchema: {}, + entry: 'src/index.ts', + events: [], + legacyAliases: [], + migrations: [], + minApiSdkVersion: '1.0.0', + name: 'Demo Plugin', + operations: [], + permissions: [], + pluginKey: 'demo-plugin', + runtime: { + maxConcurrency: 1, + memoryMb: 128, + timeoutMs: 5000, + workerType: 'node-worker', + }, + version: '0.1.0', + }; + const installation = { + id: 'installation-cleanup', + installedPath: 'D:/plugins/demo-plugin', + pluginId: '2060000000000000003', + runtimeStatus: 'stopped', + status: 'installed', + versionId: 'version-cleanup', + }; + const version = { + id: 'version-cleanup', + manifestJson: manifest, + packageHash: 'hash', + pluginId: installation.pluginId, + version: '0.1.0', + }; + const createRepository = (findOneValue?: unknown) => ({ + find: jest.fn(async () => []), + findAndCount: jest.fn(async () => [[], 0]), + findOne: jest.fn(async () => findOneValue), + save: jest.fn(async (value) => value), + update: jest.fn(async () => ({ affected: 1 })), + }); + const runtimeEventRepository = createRepository(); + runtimeEventRepository.save = jest.fn(async (value) => { + if ( + value && + typeof value === 'object' && + (value as { eventType?: string }).eventType === + 'worker-dispose-finished' + ) { + throw new Error('runtime event db unavailable'); + } + return value; + }); + const runtimeEventBatches = [ + [], + [ + { + eventType: 'worker-dispose-finished', + level: 'info', + pluginKey: 'demo-plugin', + safeSummary: { + phase: 'dispose', + }, + }, + ], + ]; + const worker = { + activate: jest.fn(async () => ({ ok: true })), + deactivate: jest.fn(async () => ({ ok: true })), + dispose: jest.fn(async () => undefined), + drainRuntimeEvents: jest.fn(() => runtimeEventBatches.shift() || []), + health: jest.fn(async () => ({ ok: true })), + load: jest.fn(async () => ({ ok: true })), + }; + const service = new (QqbotPluginPlatformService as any)( + createRepository({ + id: installation.pluginId, + pluginKey: 'demo-plugin', + }), + createRepository(version), + createRepository(installation), + createRepository(), + createRepository(), + createRepository(), + createRepository(), + createRepository(), + runtimeEventRepository, + undefined, + { + create: jest.fn(() => worker), + }, + { + setPluginActive: jest.fn(), + }, + { + setPluginActive: jest.fn(), + }, + ) as QqbotPluginPlatformService; + + await service.enableInstallation({ id: installation.id }); + await expect( + service.disableInstallation({ id: installation.id }), + ).resolves.toMatchObject({ + id: installation.id, + runtimeStatus: 'stopped', + status: 'disabled', + }); + + expect(worker.dispose).toHaveBeenCalled(); + expect( + (service as any).activeWorkers.has(installation.id), + ).toBe(false); + expect( + (service as any).activeWorkerContexts.has(installation.id), + ).toBe(false); }); it('routes enabled command and message executions through active worker runtimes', async () => { @@ -331,7 +512,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { const installation = { id: 'installation-execute', installedPath: 'D:/plugins/demo-plugin', - pluginId: 'plugin-execute', + pluginId: '2060000000000000001', runtimeStatus: 'stopped', status: 'installed', versionId: 'version-execute', @@ -350,10 +531,25 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { save: jest.fn(async (value) => value), update: jest.fn(async () => ({ affected: 1 })), }); + const runtimeEventRepository = createRepository(); + const runtimeEventBatches = [ + [ + { + eventType: 'worker-recovered', + level: 'info', + pluginKey: 'demo-plugin', + safeSummary: { + status: 'active', + }, + }, + ], + [], + ]; const worker = { activate: jest.fn(async () => ({ ok: true })), deactivate: jest.fn(async () => ({ ok: true })), dispose: jest.fn(async () => undefined), + drainRuntimeEvents: jest.fn(() => runtimeEventBatches.shift() || []), executeOperation: jest.fn(async () => ({ replyText: 'worker-ok' })), handleEvent: jest.fn(async () => true), health: jest.fn(async () => ({ ok: true })), @@ -398,7 +594,7 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { createRepository(), createRepository(), createRepository(), - createRepository(), + runtimeEventRepository, argumentParser, runtimeFactory, fallbackCommandRegistry, @@ -452,6 +648,17 @@ describe('QQBot plugin platform lifecycle runtime contract', () => { ); expect(fallbackCommandRegistry.execute).not.toHaveBeenCalled(); expect(fallbackEventRegistry.dispatchMessage).not.toHaveBeenCalled(); + expect(runtimeEventRepository.save).toHaveBeenCalledWith( + expect.objectContaining({ + eventType: 'worker-recovered', + installationId: installation.id, + level: 'info', + pluginId: installation.pluginId, + safeSummary: { + status: 'active', + }, + }), + ); }); it('does not execute workers for persisted disabled built-in plugins', async () => { diff --git a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts index d3bdfa0..2da7b18 100644 --- a/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts +++ b/test/modules/qqbot/plugin-platform/worker-runtime.spec.ts @@ -1,8 +1,13 @@ +import type { ConfigService } from '@nestjs/config'; import { + createQqbotBullmqWorkerQueueOptions, QqbotPluginRuntimeError, + QqbotPluginWorkerResponseError, QqbotPluginWorkerRuntime, + resolveQqbotPluginQueueConnection, type QqbotPluginWorkerDriver, type QqbotPluginWorkerRequest, + type QqbotPluginWorkerRequestQueue, } from '../../../../src/modules/qqbot/plugin-platform/infrastructure/integration/runtime'; import { createQqbotPluginSdk } from '../../../../src/modules/qqbot/plugin-platform/infrastructure/integration/sdk'; @@ -34,8 +39,75 @@ class RecordingDriver implements QqbotPluginWorkerDriver { } } +class RecordingRequestQueue implements QqbotPluginWorkerRequestQueue { + private previous = Promise.resolve(); + + constructor(private readonly driver: QqbotPluginWorkerDriver) {} + + async close() { + await this.driver.dispose(); + } + + request(message: QqbotPluginWorkerRequest): Promise { + const run = this.previous + .catch(() => undefined) + .then(() => this.driver.request(message)); + this.previous = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + async reset() { + await this.driver.dispose(); + } +} + +class GenerationAwareRequestQueue implements QqbotPluginWorkerRequestQueue { + private generation = 0; + private previous = Promise.resolve(); + + constructor(private readonly driver: QqbotPluginWorkerDriver) {} + + async close() { + await this.driver.dispose(); + } + + request(message: QqbotPluginWorkerRequest): Promise { + const generation = this.generation; + const run = this.previous + .catch(() => undefined) + .then(async () => { + if (generation !== this.generation) { + const error = new Error('stale worker queue request'); + error.name = 'QqbotPluginWorkerStaleRequestError'; + throw error; + } + try { + return await this.driver.request(message); + } catch (error) { + if (!(error instanceof QqbotPluginWorkerResponseError)) { + this.generation += 1; + } + throw error; + } + }); + this.previous = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + async reset() { + this.generation += 1; + await this.driver.dispose(); + } +} + const createRuntime = (driver = new RecordingDriver()) => { - const runtime = new QqbotPluginWorkerRuntime(driver, { + const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { defaultTimeoutMs: 50, installationId: 'install-1', pluginKey: 'demo-plugin', @@ -45,6 +117,190 @@ const createRuntime = (driver = new RecordingDriver()) => { }; describe('QQBot plugin worker runtime', () => { + it('serializes concurrent worker execution requests for one plugin runtime', async () => { + const releaseFirst = createDeferred(); + const releaseSecond = createDeferred(); + let activeRequests = 0; + let maxActiveRequests = 0; + const requestOrder: string[] = []; + const driver: QqbotPluginWorkerDriver = { + dispose: jest.fn(async () => undefined), + request: jest.fn(async (message) => { + activeRequests += 1; + maxActiveRequests = Math.max(maxActiveRequests, activeRequests); + requestOrder.push(message.operationId || message.type); + + if (message.operationId === 'op-1') { + await releaseFirst.promise; + } + if (message.operationId === 'op-2') { + await releaseSecond.promise; + } + + activeRequests -= 1; + return { + ok: true, + operationId: message.operationId, + }; + }), + }; + const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { + defaultTimeoutMs: 100, + installationId: 'install-serial', + pluginKey: 'demo-plugin', + }); + + const first = runtime.executeOperation({ + input: { text: 'first' }, + operationId: 'op-1', + operationKey: 'demo-plugin.echo', + }); + const second = runtime.executeOperation({ + input: { text: 'second' }, + operationId: 'op-2', + operationKey: 'demo-plugin.echo', + }); + + try { + await waitUntil(() => requestOrder.includes('op-1')); + expect(requestOrder).toEqual(['op-1']); + + releaseFirst.resolve(); + await expect(first).resolves.toMatchObject({ operationId: 'op-1' }); + + await waitUntil(() => requestOrder.includes('op-2')); + expect(maxActiveRequests).toBe(1); + releaseSecond.resolve(); + await expect(second).resolves.toMatchObject({ operationId: 'op-2' }); + } finally { + releaseFirst.resolve(); + releaseSecond.resolve(); + await Promise.allSettled([first, second]); + } + }); + + it('recovers a failed worker by reloading and activating it before the next request', async () => { + const requestTypes: string[] = []; + const driver: QqbotPluginWorkerDriver = { + dispose: jest.fn(async () => undefined), + request: jest.fn(async (message) => { + requestTypes.push(message.type); + if (message.operationId === 'op-crash') { + throw new Error('worker crashed'); + } + return { + ok: true, + operationId: message.operationId, + type: message.type, + }; + }), + }; + const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { + defaultTimeoutMs: 50, + installationId: 'install-recover', + pluginKey: 'demo-plugin', + }); + const manifest = { + entry: 'src/index.ts', + pluginKey: 'demo-plugin', + version: '0.1.0', + }; + + await runtime.load(manifest); + await runtime.activate(); + await expect( + runtime.executeOperation({ + input: { text: 'boom' }, + operationId: 'op-crash', + operationKey: 'demo-plugin.echo', + }), + ).rejects.toMatchObject({ + code: 'PLUGIN_WORKER_CRASH', + }); + expect(runtime.status).toBe('failed'); + + await expect( + runtime.executeOperation({ + input: { text: 'after crash' }, + operationId: 'op-after-crash', + operationKey: 'demo-plugin.echo', + }), + ).resolves.toMatchObject({ operationId: 'op-after-crash' }); + + expect(driver.dispose).toHaveBeenCalled(); + expect(requestTypes).toEqual([ + 'load', + 'activate', + 'executeOperation', + 'load', + 'activate', + 'executeOperation', + ]); + expect(runtime.status).toBe('active'); + }); + + it('recovers before retrying requests that were already queued when a worker crashed', async () => { + const requestTypes: string[] = []; + const driver: QqbotPluginWorkerDriver = { + dispose: jest.fn(async () => undefined), + request: jest.fn(async (message) => { + requestTypes.push(message.operationId || message.type); + if (message.operationId === 'op-crash') { + throw new Error('worker crashed'); + } + return { + ok: true, + operationId: message.operationId, + type: message.type, + }; + }), + }; + const runtime = new QqbotPluginWorkerRuntime( + new GenerationAwareRequestQueue(driver), + { + defaultTimeoutMs: 50, + installationId: 'install-stale-retry', + pluginKey: 'demo-plugin', + }, + ); + + await runtime.load({ + entry: 'src/index.ts', + pluginKey: 'demo-plugin', + version: '0.1.0', + }); + await runtime.activate(); + + const first = runtime + .executeOperation({ + input: { text: 'boom' }, + operationId: 'op-crash', + operationKey: 'demo-plugin.echo', + }) + .catch((error) => error); + const second = runtime.executeOperation({ + input: { text: 'after crash' }, + operationId: 'op-after-crash', + operationKey: 'demo-plugin.echo', + }); + + await expect(first).resolves.toMatchObject({ + code: 'PLUGIN_WORKER_CRASH', + }); + await expect(second).resolves.toMatchObject({ + operationId: 'op-after-crash', + }); + expect(requestTypes).toEqual([ + 'load', + 'activate', + 'op-crash', + 'load', + 'activate', + 'op-after-crash', + ]); + expect(runtime.status).toBe('active'); + }); + it('sends lifecycle and execution RPC messages with correlation IDs and safe input summaries', async () => { const { driver, runtime } = createRuntime(); driver.responses.set('executeOperation', { replyText: 'ok' }); @@ -143,6 +399,42 @@ describe('QQBot plugin worker runtime', () => { ]); }); + it('keeps plugin response errors from poisoning the worker runtime', async () => { + const driver = new RecordingDriver( + new Map([ + [ + 'executeOperation', + new QqbotPluginWorkerResponseError({ + message: '业务参数错误', + name: 'PluginCommandError', + }), + ], + ]), + ); + const { runtime } = createRuntime(driver); + + await runtime.load({ + entry: 'src/index.ts', + pluginKey: 'demo-plugin', + version: '0.1.0', + }); + await runtime.activate(); + await expect( + runtime.executeOperation({ + input: { text: 'bad input' }, + operationId: 'op-plugin-error', + operationKey: 'demo-plugin.echo', + }), + ).rejects.toMatchObject({ + message: '业务参数错误', + name: 'PluginCommandError', + }); + + expect(runtime.status).toBe('active'); + expect(driver.disposed).toBe(false); + expect(runtime.listRuntimeEvents()).toEqual([]); + }); + it('times out slow worker calls and disposes the driver boundary', async () => { const driver: QqbotPluginWorkerDriver = { dispose: jest.fn(async () => undefined), @@ -150,7 +442,7 @@ describe('QQBot plugin worker runtime', () => { () => new Promise((resolve) => setTimeout(resolve, 1000)), ), }; - const runtime = new QqbotPluginWorkerRuntime(driver, { + const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { defaultTimeoutMs: 5, installationId: 'install-timeout', pluginKey: 'demo-plugin', @@ -181,7 +473,7 @@ describe('QQBot plugin worker runtime', () => { dispose: jest.fn(async () => undefined), request: jest.fn(async () => ({ ok: true })), }; - const runtime = new QqbotPluginWorkerRuntime(driver, { + const runtime = new QqbotPluginWorkerRuntime(new RecordingRequestQueue(driver), { defaultTimeoutMs: 5, installationId: 'install-success', pluginKey: 'demo-plugin', @@ -201,6 +493,29 @@ describe('QQBot plugin worker runtime', () => { }); }); +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((innerResolve, innerReject) => { + resolve = innerResolve; + reject = innerReject; + }); + return { promise, reject, resolve }; +} + +async function waitUntil( + condition: () => boolean, + timeoutMs = 100, +): Promise { + const deadline = Date.now() + timeoutMs; + while (!condition()) { + if (Date.now() > deadline) { + throw new Error('condition was not met before timeout'); + } + await new Promise((resolve) => setTimeout(resolve, 1)); + } +} + describe('QQBot plugin SDK contract', () => { it('exposes only host-controlled capabilities to plugin code', () => { const sdk = createQqbotPluginSdk({ @@ -248,3 +563,49 @@ describe('QQBot plugin SDK contract', () => { expect('repository' in sdk).toBe(false); }); }); + +describe('QQBot plugin worker queue config', () => { + it('requires an explicit Redis host for plugin worker queues', () => { + const configService = createConfigService({}); + + expect(() => resolveQqbotPluginQueueConnection(configService)).toThrow( + 'QQBot 插件队列缺少 Redis 主机配置', + ); + }); + + it('uses safe numeric defaults when optional Redis values are blank', () => { + const options = createQqbotBullmqWorkerQueueOptions( + createConfigService({ + QQBOT_PLUGIN_QUEUE_REDIS_DB: '', + QQBOT_PLUGIN_QUEUE_REDIS_HOST: 'redis.local', + QQBOT_PLUGIN_QUEUE_REDIS_PORT: '', + QQBOT_PLUGIN_QUEUE_REDIS_PREFIX: '', + QQBOT_PLUGIN_QUEUE_REMOVE_ON_FAIL: '', + QQBOT_PLUGIN_QUEUE_WAIT_BUFFER_MS: '', + }), + 'bangdream', + 'install-1', + ); + + expect(options).toMatchObject({ + connection: { + db: 0, + host: 'redis.local', + port: 6379, + }, + installationId: 'install-1', + pluginKey: 'bangdream', + prefix: 'kt:qqbot:plugin-worker', + removeOnFailCount: 100, + waitUntilFinishedBufferMs: 5_000, + workerInstanceId: expect.any(String), + }); + expect(options.workerInstanceId).toContain(':'); + }); +}); + +function createConfigService(values: Record): ConfigService { + return { + get: (key: string) => values[key] as T, + } as ConfigService; +}