diff --git a/.env.example b/.env.example index 18a565c..69ad361 100644 --- a/.env.example +++ b/.env.example @@ -10,7 +10,7 @@ DATABASE_URL=mysql://devops_user:devops_password@127.0.0.1:3306/devops_platform USE_DATABASE_READS=false PRISMA_CONNECT_ON_BOOT=false -# Redis and queues, reserved for async polling/notification workers. +# Redis and queues. Used by health checks and BullMQ-backed notification outbox retry. REDIS_URL=redis://127.0.0.1:6379/0 # Jenkins integration, server-side only. @@ -34,6 +34,11 @@ NOTIFICATION_PROVIDER=replace-with-provider NOTIFICATION_WEBHOOK_URL=https://notification.example.internal/webhook/replace-with-secret-manager-value FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/replace-with-secret-manager-value WECOM_WEBHOOK_URL=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=replace-with-secret-manager-value +NOTIFICATION_OUTBOX_RETRY_ENABLED=false +NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS=30000 +NOTIFICATION_OUTBOX_RETRY_DELAY_MS=60000 +NOTIFICATION_OUTBOX_MAX_ATTEMPTS=3 +NOTIFICATION_OUTBOX_BATCH_SIZE=10 # LLM proxy configuration, server-side only. LLM_BASE_URL=https://llm.example.internal/v1 diff --git a/README.md b/README.md index ea2d3cd..0d21e11 100644 --- a/README.md +++ b/README.md @@ -103,10 +103,11 @@ pnpm prisma:seed - `DATABASE_URL`：MySQL 连接地址。 - `USE_DATABASE_READS`：是否启用 Prisma 数据读写。 - `PRISMA_CONNECT_ON_BOOT`：启动时是否主动连接数据库。 -- `REDIS_URL`：Redis 连接地址，`/health` 会执行真实 PING；后续 BullMQ worker 复用该连接边界。 +- `REDIS_URL`：Redis 连接地址，`/health` 会执行真实 PING，通知 outbox 重试会优先使用 BullMQ 调度。 - `JENKINS_BASE_URL`、`JENKINS_USERNAME`、`JENKINS_API_TOKEN`：Jenkins 集成配置。 - `GITEA_BASE_URL`、`GITEA_TOKEN`、`GITEA_WEBHOOK_SECRET`、`GITEA_WEBHOOK_RELAY_BASE_URL`：Gitea 集成与 Jenkins relay 诊断配置。 - `NOTIFICATION_PROVIDER`、`WECOM_WEBHOOK_URL`、`FEISHU_WEBHOOK_URL`、`NOTIFICATION_WEBHOOK_URL`：通知集成配置。 +- `NOTIFICATION_OUTBOX_RETRY_ENABLED`、`NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS`、`NOTIFICATION_OUTBOX_RETRY_DELAY_MS`、`NOTIFICATION_OUTBOX_MAX_ATTEMPTS`、`NOTIFICATION_OUTBOX_BATCH_SIZE`：通知 outbox 自动重试配置；配置 `REDIS_URL` 时使用 BullMQ 调度，否则降级为进程内定时器。 - `LLM_BASE_URL`、`LLM_API_KEY`、`LLM_MODEL`：Agent 配置的环境变量后备值。 ## Agent 边界 diff --git a/RTK.md b/RTK.md index 28b03d5..755b347 100644 --- a/RTK.md +++ b/RTK.md @@ -19,7 +19,7 @@ - `deploy-runs`：发布记录 API、状态模型和 BPMN 步骤摘要。 - `integrations/jenkins`：Jenkins API 客户端边界。 - `integrations/gitea`：Gitea API 客户端边界。 -- `integrations/redis`：Redis PING 健康检查边界，后续 BullMQ worker 复用该模块。 +- `integrations/redis`：Redis PING 健康检查和 BullMQ 连接参数边界，通知 outbox 调度复用该模块。 - `notifications/wecom`：企微、飞书和通用 webhook 通知边界。 - `agent-config`：Agent 系统级配置、密钥加密和连接测试。 - `agent`：只面向运维工作流的 LLM 代理边界。 diff --git a/package.json b/package.json index de693e8..3e1424d 100644 --- a/package.json +++ b/package.json @@ -30,6 +30,7 @@ "@nestjs/platform-express": "^11.1.6", "@nestjs/swagger": "^11.2.0", "@prisma/client": "^6.14.0", + "bullmq": "^5.78.0", "ioredis": "^5.11.1", "pino": "^10.3.1", "reflect-metadata": "^0.2.2", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 423f3f0..a5d0707 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -26,6 +26,9 @@ importers: '@prisma/client': specifier: ^6.14.0 version: 6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3) + bullmq: + specifier: ^5.78.0 + version: 5.78.0 ioredis: specifier: ^5.11.1 version: 5.11.1 @@ -513,6 +516,9 @@ packages: '@ioredis/commands@1.10.0': resolution: {integrity: sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==} + '@ioredis/commands@1.5.1': + resolution: {integrity: sha512-JH8ZL/ywcJyR9MmJ5BNqZllXNZQqQbnVZOqpPQqE1vHiFgAw4NHbvE0FOduNU8IX9babitBT46571OnPTT0Zcw==} + '@isaacs/cliui@8.0.2': resolution: {integrity: sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==} engines: {node: '>=12'} @@ -636,6 +642,36 @@ 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] + '@napi-rs/wasm-runtime@1.1.5': resolution: {integrity: sha512-AWPoBRJ9tsnVhor4sjO7rkni+7p+2IAEFj6cx06UgP10jkQHqay/36uRV/bFkgrh18D9vb4cr8Q0Pthskgzy+Q==} peerDependencies: @@ -1288,6 +1324,15 @@ packages: buffer@5.7.1: resolution: {integrity: sha512-EHcyIPBQ4BSGlvjB16k5KgAJ27CIsHY/2JBmCRReo48y9rQ3MaUzWX3KVlBa4U7MyX02HdVj0K7C3WaB3ju7FQ==} + bullmq@5.78.0: + resolution: {integrity: sha512-tT9jJmbobk9ueEfFc22egLmgwCcMGgOjZ5Y1cvgczBPv1JUmC7iHQVbQtqku2YBE5dE9uzdVpxIrBvL/YAjGwA==} + 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'} @@ -1465,6 +1510,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-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -1514,6 +1563,10 @@ packages: destr@2.0.5: resolution: {integrity: sha512-ugFTXCtDZunbzasqBxrK93Ik/DRYsO6S/fedkWEMKqt04xZ4csmnmwGDBAb07QWNaGMAmnTIemsYZCksjATwsA==} + detect-libc@2.1.2: + resolution: {integrity: sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==} + engines: {node: '>=8'} + detect-newline@3.1.0: resolution: {integrity: sha512-TLz+x/vEXm/Y7P7wn1EJFNLxYpUD4TgMosxY6fAVJUnJMbupHBOncxyWUG9OpTaH9EBD7uFI5LfEgmMOc54DsA==} engines: {node: '>=8'} @@ -1914,6 +1967,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'} + ioredis@5.11.1: resolution: {integrity: sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==} engines: {node: '>=12.22.0'} @@ -2198,6 +2255,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==} @@ -2221,6 +2284,10 @@ packages: lru-cache@5.1.1: resolution: {integrity: sha512-KpNARQA3Iwv+jTA0utUVVbrh+Jlrr1Fv0e56GGzAFOXN7dk/FviaDW8LHmK52DlcH4WP2n6gI8vN1aesBFgo9w==} + 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==} @@ -2298,6 +2365,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'} @@ -2330,6 +2404,10 @@ packages: node-fetch-native@1.6.7: resolution: {integrity: sha512-g9yhqoedzIUm0nTnTqAQvueMPVOuIY16bqgAJJC8XOOubYFNwz6IER9qs0Gq2Xd0+CecCKFjtdDTMA4u4xG06Q==} + 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==} @@ -2634,6 +2712,11 @@ packages: resolution: {integrity: sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA==} hasBin: true + semver@7.8.0: + resolution: {integrity: sha512-AcM7dV/5ul4EekoQ29Agm5vri8JNqRyj39o0qpX6vDF2GZrtutZl5RwgD1XnZjiTAfncsJhMI48QQH3sN87YNA==} + engines: {node: '>=10'} + hasBin: true + semver@7.8.4: resolution: {integrity: sha512-rUCObTnP32Q08R2uuIrt7r9PlEonuTmtuXYcW6s5kjdlj3xbnwe+21yXptAUYcMAABLkYYTtnmzb3w3EDZfueA==} engines: {node: '>=10'} @@ -3575,6 +3658,8 @@ snapshots: '@ioredis/commands@1.10.0': {} + '@ioredis/commands@1.5.1': {} + '@isaacs/cliui@8.0.2': dependencies: string-width: 5.1.2 @@ -3805,6 +3890,24 @@ 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 + '@napi-rs/wasm-runtime@1.1.5(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)': dependencies: '@emnapi/core': 1.10.0 @@ -4526,6 +4629,17 @@ snapshots: base64-js: 1.5.1 ieee754: 1.2.1 + bullmq@5.78.0: + dependencies: + cron-parser: 4.9.0 + ioredis: 5.10.1 + msgpackr: 2.0.2 + node-abort-controller: 3.1.1 + semver: 7.8.0 + tslib: 2.8.1 + transitivePeerDependencies: + - supports-color + busboy@1.6.0: dependencies: streamsearch: 1.1.0 @@ -4674,6 +4788,10 @@ snapshots: create-require@1.1.1: {} + cron-parser@4.9.0: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -4704,6 +4822,9 @@ snapshots: destr@2.0.5: {} + detect-libc@2.1.2: + optional: true + detect-newline@3.1.0: {} diff@4.0.4: {} @@ -5148,6 +5269,20 @@ snapshots: inherits@2.0.4: {} + ioredis@5.10.1: + dependencies: + '@ioredis/commands': 1.5.1 + cluster-key-slot: 1.1.1 + 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 + ioredis@5.11.1: dependencies: '@ioredis/commands': 1.10.0 @@ -5605,6 +5740,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: {} @@ -5624,6 +5763,8 @@ snapshots: dependencies: yallist: 3.1.1 + luxon@3.7.2: {} + magic-string@0.30.17: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -5684,6 +5825,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 @@ -5709,6 +5866,11 @@ snapshots: node-fetch-native@1.6.7: {} + 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: {} @@ -5996,6 +6158,8 @@ snapshots: semver@6.3.1: {} + semver@7.8.0: {} + semver@7.8.4: {} send@1.2.1: diff --git a/src/integrations/redis/bullmq-connection.factory.ts b/src/integrations/redis/bullmq-connection.factory.ts new file mode 100644 index 0000000..b64ba32 --- /dev/null +++ b/src/integrations/redis/bullmq-connection.factory.ts @@ -0,0 +1,33 @@ +/** + * BullMQ 连接工厂统一封装 Redis URL 和 Worker 连接约束，避免业务模块散落连接参数。 + */ +import { Injectable } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { ConnectionOptions } from 'bullmq'; +import { EnvConfig } from '../../config/env.schema'; + +@Injectable() +export class BullMqConnectionFactory { + constructor(private readonly config: ConfigService) {} + + /** + * 创建 BullMQ Queue/Worker 共用连接参数；Worker 阻塞连接要求请求重试不设上限。 + */ + createConnectionOptions( + connectionName: string, + ): ConnectionOptions | undefined { + const redisUrl = this.config.get('REDIS_URL', { infer: true }); + + if (!redisUrl) { + return undefined; + } + + return { + url: redisUrl, + connectionName, + connectTimeout: 2000, + maxRetriesPerRequest: null, + retryStrategy: (attempt) => Math.min(attempt * 200, 5000), + }; + } +} diff --git a/src/integrations/redis/redis.module.ts b/src/integrations/redis/redis.module.ts index d00b614..16b0994 100644 --- a/src/integrations/redis/redis.module.ts +++ b/src/integrations/redis/redis.module.ts @@ -1,8 +1,9 @@ import { Module } from '@nestjs/common'; +import { BullMqConnectionFactory } from './bullmq-connection.factory'; import { RedisClient } from './redis.client'; @Module({ - providers: [RedisClient], - exports: [RedisClient], + providers: [BullMqConnectionFactory, RedisClient], + exports: [BullMqConnectionFactory, RedisClient], }) export class RedisModule {} diff --git a/src/notifications/wecom/notification-outbox-retry.service.ts b/src/notifications/wecom/notification-outbox-retry.service.ts index 97bfb91..7ee5042 100644 --- a/src/notifications/wecom/notification-outbox-retry.service.ts +++ b/src/notifications/wecom/notification-outbox-retry.service.ts @@ -1,30 +1,52 @@ /** - * 通知 outbox 重试器复用当前 provider 投递边界，并避免空扫描刷屏审计。 - * 后续迁移到 BullMQ 时保留“有候选才记录成功 tick”的可观测边界。 + * 通知 outbox 重试器以 BullMQ 作为 Redis 调度层，并保留内存定时器降级路径。 + * 真实候选查询、投递和审计仍集中在 runOnce，避免队列层持有业务状态。 */ import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; +import { Queue, Worker } from 'bullmq'; import { AuditService } from '../../audit/audit.service'; import { redactSensitive } from '../../common/security/redact-sensitive'; import { EnvConfig } from '../../config/env.schema'; +import { BullMqConnectionFactory } from '../../integrations/redis/bullmq-connection.factory'; import { NotificationOutboxService } from '../outbox/notification-outbox.service'; import { DeployNotificationService } from './deploy-notification.service'; +type NotificationOutboxRetryJobData = { + source: 'bullmq'; + scheduledAt: string; +}; + +const retryQueueName = 'notification-outbox-retry'; +const retryJobName = 'retry-tick'; +const retrySchedulerId = 'notification-outbox-retry-scheduler'; +const bullMqPrefix = 'devops-platform'; +const bullMqStartupTimeoutMs = 5000; + @Injectable() export class NotificationOutboxRetryService implements OnModuleInit, OnModuleDestroy { private timer: ReturnType | undefined; + private queue: + | Queue + | undefined; + private worker: + | Worker + | undefined; private running = false; + private bullMqFallbackRecorded = false; + private bullMqRuntimeIssueRecorded = false; constructor( private readonly config: ConfigService, + private readonly bullMqConnectionFactory: BullMqConnectionFactory, private readonly notificationOutboxService: NotificationOutboxService, private readonly deployNotificationService: DeployNotificationService, private readonly auditService: AuditService, ) {} - onModuleInit(): void { + async onModuleInit(): Promise { if ( !this.config.get('NOTIFICATION_OUTBOX_RETRY_ENABLED', { infer: true }) ) { @@ -36,19 +58,120 @@ export class NotificationOutboxRetryService { infer: true }, ); + await this.startBullMqScheduler(intervalMs); + } + + async onModuleDestroy(): Promise { + this.stopMemoryScheduler(); + await this.closeBullMqScheduler(false); + } + + private async startBullMqScheduler(intervalMs: number): Promise { + const connection = this.bullMqConnectionFactory.createConnectionOptions( + 'notification-outbox-retry', + ); + + if (!connection) { + this.startMemoryScheduler(intervalMs); + return; + } + + try { + this.queue = new Queue( + retryQueueName, + { + connection, + prefix: bullMqPrefix, + defaultJobOptions: { + removeOnComplete: { count: 20 }, + removeOnFail: { count: 20 }, + }, + }, + ); + this.worker = new Worker( + retryQueueName, + async () => { + await this.runOnce(); + }, + { + connection, + prefix: bullMqPrefix, + concurrency: 1, + removeOnComplete: { count: 20 }, + removeOnFail: { count: 20 }, + }, + ); + + this.queue.on('error', (error) => { + void this.recordBullMqIssue('queue', error, 'bullmq_recovering'); + }); + this.worker.on('error', (error) => { + void this.recordBullMqIssue('worker', error, 'bullmq_recovering'); + }); + this.worker.on('failed', (_job, error) => { + void this.recordBullMqIssue('job', error, 'bullmq_recovering'); + }); + + await this.withStartupTimeout( + Promise.all([ + this.queue.waitUntilReady(), + this.worker.waitUntilReady(), + ]), + ); + await this.queue.upsertJobScheduler( + retrySchedulerId, + { every: intervalMs }, + { + name: retryJobName, + data: { + source: 'bullmq', + scheduledAt: new Date().toISOString(), + }, + opts: { + removeOnComplete: { count: 20 }, + removeOnFail: { count: 20 }, + }, + }, + ); + void this.runOnce(); + } catch (error) { + await this.closeBullMqScheduler(true); + await this.recordBullMqIssue('startup', error, 'memory_timer'); + this.startMemoryScheduler(intervalMs); + } + } + + private startMemoryScheduler(intervalMs: number): void { + if (this.timer) { + return; + } + this.timer = setInterval(() => { void this.runOnce(); }, intervalMs); void this.runOnce(); } - onModuleDestroy(): void { + private stopMemoryScheduler(): void { if (this.timer) { clearInterval(this.timer); this.timer = undefined; } } + private async closeBullMqScheduler(forceWorkerClose: boolean): Promise { + const worker = this.worker; + const queue = this.queue; + + this.worker = undefined; + this.queue = undefined; + + await Promise.all([ + worker?.close(forceWorkerClose).catch(() => undefined), + queue?.close().catch(() => undefined), + ]); + } + async runOnce(): Promise { if (this.running) { return; @@ -63,13 +186,12 @@ export class NotificationOutboxRetryService const batchSize = this.config.get('NOTIFICATION_OUTBOX_BATCH_SIZE', { infer: true, }); - const messages = await this.notificationOutboxService.listRetryableMessages( - { + const messages = + await this.notificationOutboxService.listRetryableMessages({ now: new Date(), maxAttempts, take: batchSize, - }, - ); + }); let sentCount = 0; let failedCount = 0; let skippedCount = 0; @@ -116,4 +238,57 @@ export class NotificationOutboxRetryService this.running = false; } } + + private async withStartupTimeout(promise: Promise): Promise { + return Promise.race([ + promise, + new Promise((_resolve, reject) => { + setTimeout( + () => reject(new Error('BullMQ 调度器启动超时')), + bullMqStartupTimeoutMs, + ); + }), + ]); + } + + private async recordBullMqIssue( + stage: 'startup' | 'queue' | 'worker' | 'job', + error: unknown, + mode: 'memory_timer' | 'bullmq_recovering', + ): Promise { + const recorded = + mode === 'memory_timer' + ? this.bullMqFallbackRecorded + : this.bullMqRuntimeIssueRecorded; + + if (recorded) { + return; + } + + if (mode === 'memory_timer') { + this.bullMqFallbackRecorded = true; + } else { + this.bullMqRuntimeIssueRecorded = true; + } + + const message = + error instanceof Error ? error.message : 'BullMQ 通知重试调度异常'; + + try { + await this.auditService.record({ + action: + mode === 'memory_timer' + ? 'NOTIFICATION_OUTBOX_RETRY_SCHEDULER_FALLBACK' + : 'NOTIFICATION_OUTBOX_RETRY_SCHEDULER_DEGRADED', + resourceType: 'notification_outbox', + after: { + stage, + mode, + message: redactSensitive({ message }).message, + }, + }); + } catch { + // 启动早期数据库可能尚不可用，调度降级不能被审计写入失败阻断。 + } + } } diff --git a/src/notifications/wecom/wecom.module.ts b/src/notifications/wecom/wecom.module.ts index d0a6a5b..848a9f8 100644 --- a/src/notifications/wecom/wecom.module.ts +++ b/src/notifications/wecom/wecom.module.ts @@ -2,6 +2,7 @@ * 通知模块集中注册 provider 客户端、发布通知服务和 outbox 能力。 */ import { Module } from '@nestjs/common'; +import { RedisModule } from '../../integrations/redis/redis.module'; import { NotificationOutboxModule } from '../outbox/notification-outbox.module'; import { DeployNotificationService } from './deploy-notification.service'; import { NotificationOutboxManualRetryController } from './notification-outbox-manual-retry.controller'; @@ -10,7 +11,7 @@ import { NotificationOutboxRetryService } from './notification-outbox-retry.serv import { WeComClient } from './wecom.client'; @Module({ - imports: [NotificationOutboxModule], + imports: [NotificationOutboxModule, RedisModule], controllers: [NotificationOutboxManualRetryController], providers: [ WeComClient,