diff --git a/.env.example b/.env.example index 69ad361..d81a59e 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. Used by health checks and BullMQ-backed notification outbox retry. +# Redis and queues. Used by health checks, Jenkins auto sync, and notification outbox retry. REDIS_URL=redis://127.0.0.1:6379/0 # Jenkins integration, server-side only. diff --git a/README.md b/README.md index 0d21e11..088fd41 100644 --- a/README.md +++ b/README.md @@ -103,7 +103,7 @@ pnpm prisma:seed - `DATABASE_URL`:MySQL 连接地址。 - `USE_DATABASE_READS`:是否启用 Prisma 数据读写。 - `PRISMA_CONNECT_ON_BOOT`:启动时是否主动连接数据库。 -- `REDIS_URL`:Redis 连接地址,`/health` 会执行真实 PING,通知 outbox 重试会优先使用 BullMQ 调度。 +- `REDIS_URL`:Redis 连接地址,`/health` 会执行真实 PING,Jenkins 自动同步和通知 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`:通知集成配置。 diff --git a/RTK.md b/RTK.md index 755b347..712e9f4 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 连接参数边界,通知 outbox 调度复用该模块。 +- `integrations/redis`:Redis PING 健康检查和 BullMQ 连接参数边界,Jenkins 自动同步与通知 outbox 调度复用该模块。 - `notifications/wecom`:企微、飞书和通用 webhook 通知边界。 - `agent-config`:Agent 系统级配置、密钥加密和连接测试。 - `agent`:只面向运维工作流的 LLM 代理边界。 diff --git a/src/deploy-runs/deploy-runs.module.ts b/src/deploy-runs/deploy-runs.module.ts index 5163b3a..15c44aa 100644 --- a/src/deploy-runs/deploy-runs.module.ts +++ b/src/deploy-runs/deploy-runs.module.ts @@ -1,7 +1,11 @@ +/** + * 发布运行模块集中装配发布执行、Jenkins 同步、项目权限和通知依赖。 + */ import { Module } from '@nestjs/common'; import { AuditModule } from '../audit/audit.module'; import { AuthModule } from '../auth/auth.module'; import { JenkinsModule } from '../integrations/jenkins/jenkins.module'; +import { RedisModule } from '../integrations/redis/redis.module'; import { WeComModule } from '../notifications/wecom/wecom.module'; import { ProjectsModule } from '../projects/projects.module'; import { DeployExecutionService } from './deploy-execution.service'; @@ -11,7 +15,14 @@ import { DeployRunsService } from './deploy-runs.service'; import { JenkinsSyncSchedulerService } from './jenkins-sync-scheduler.service'; @Module({ - imports: [AuditModule, AuthModule, JenkinsModule, ProjectsModule, WeComModule], + imports: [ + AuditModule, + AuthModule, + JenkinsModule, + ProjectsModule, + RedisModule, + WeComModule, + ], controllers: [DeployRunsController], providers: [ DeployExecutionService, diff --git a/src/deploy-runs/jenkins-sync-scheduler.service.spec.ts b/src/deploy-runs/jenkins-sync-scheduler.service.spec.ts index 0e6fbd2..69c5e92 100644 --- a/src/deploy-runs/jenkins-sync-scheduler.service.spec.ts +++ b/src/deploy-runs/jenkins-sync-scheduler.service.spec.ts @@ -1,3 +1,6 @@ +/** + * Jenkins 自动同步调度器测试覆盖关闭、成功聚合和失败脱敏审计。 + */ import { ConfigService } from '@nestjs/config'; import { AuditService } from '../audit/audit.service'; import { EnvConfig } from '../config/env.schema'; @@ -23,7 +26,7 @@ describe('JenkinsSyncSchedulerService', () => { createAudit(), ); - service.onModuleInit(); + void service.onModuleInit(); jest.advanceTimersByTime(3000); expect(sync.syncJenkinsRuns).not.toHaveBeenCalled(); @@ -89,7 +92,9 @@ describe('JenkinsSyncSchedulerService', () => { }); function createConfig( - values: Partial>, + values: Partial< + Record + >, ): ConfigService { return { get: jest.fn((key: keyof EnvConfig) => values[key]), diff --git a/src/deploy-runs/jenkins-sync-scheduler.service.ts b/src/deploy-runs/jenkins-sync-scheduler.service.ts index 88f1dde..f65b8c9 100644 --- a/src/deploy-runs/jenkins-sync-scheduler.service.ts +++ b/src/deploy-runs/jenkins-sync-scheduler.service.ts @@ -1,28 +1,55 @@ -import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +/** + * Jenkins 自动同步调度服务封装 BullMQ 周期任务和进程内定时器降级路径。 + */ +import { + Injectable, + OnModuleDestroy, + OnModuleInit, + Optional, +} 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 { DeployRunsService } from './deploy-runs.service'; +type JenkinsAutoSyncJobData = { + source: 'bullmq'; + scheduledAt: string; +}; + +const jenkinsSyncQueueName = 'jenkins-auto-sync'; +const jenkinsSyncJobName = 'sync-tick'; +const jenkinsSyncSchedulerId = 'jenkins-auto-sync-scheduler'; +const bullMqPrefix = 'devops-platform'; +const bullMqStartupTimeoutMs = 5000; + /** - * 轻量自动同步器,先复用内存发布单和现有 Jenkins 同步逻辑。 - * 后续接 Redis/BullMQ 后,这里会迁移成队列 worker。 + * Jenkins 自动同步器优先用 BullMQ 周期调度,Redis 不可用时降级进程内定时器。 + * 真正的发布单扫描和状态推进仍复用 DeployRunsService,避免队列层持有业务状态。 */ @Injectable() export class JenkinsSyncSchedulerService 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 deployRunsService: DeployRunsService, private readonly auditService: AuditService, + @Optional() + private readonly bullMqConnectionFactory?: BullMqConnectionFactory, ) {} - onModuleInit(): void { + async onModuleInit(): Promise { if (!this.config.get('JENKINS_AUTO_SYNC_ENABLED', { infer: true })) { return; } @@ -31,19 +58,121 @@ export class JenkinsSyncSchedulerService 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( + 'jenkins-auto-sync', + ); + + if (!connection) { + this.startMemoryScheduler(intervalMs); + return; + } + + try { + this.queue = new Queue( + jenkinsSyncQueueName, + { + connection, + prefix: bullMqPrefix, + defaultJobOptions: { + removeOnComplete: { count: 20 }, + removeOnFail: { count: 20 }, + }, + }, + ); + this.worker = new Worker( + jenkinsSyncQueueName, + 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( + jenkinsSyncSchedulerId, + { every: intervalMs }, + { + name: jenkinsSyncJobName, + 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; @@ -78,4 +207,57 @@ export class JenkinsSyncSchedulerService this.running = false; } } + + private async withStartupTimeout(promise: Promise): Promise { + return Promise.race([ + promise, + new Promise((_resolve, reject) => { + setTimeout( + () => reject(new Error('BullMQ Jenkins 同步调度器启动超时')), + 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 Jenkins 同步调度异常'; + + try { + await this.auditService.record({ + action: + mode === 'memory_timer' + ? 'JENKINS_AUTO_SYNC_SCHEDULER_FALLBACK' + : 'JENKINS_AUTO_SYNC_SCHEDULER_DEGRADED', + resourceType: 'deploy_run', + after: { + stage, + mode, + message: redactSensitive({ message }).message, + }, + }); + } catch { + // 启动早期数据库可能尚不可用,调度降级不能被审计写入失败阻断。 + } + } } diff --git a/src/settings/settings.service.ts b/src/settings/settings.service.ts index e79b5cc..b33e607 100644 --- a/src/settings/settings.service.ts +++ b/src/settings/settings.service.ts @@ -1,3 +1,6 @@ +/** + * 系统配置服务只返回集成变量名和配置状态,禁止暴露任何密钥原文。 + */ import { Injectable } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { EnvConfig } from '../config/env.schema'; @@ -32,16 +35,20 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [ required: ['JENKINS_BASE_URL', 'JENKINS_USERNAME', 'JENKINS_API_TOKEN'], optional: ['JENKINS_AUTO_SYNC_ENABLED', 'JENKINS_AUTO_SYNC_INTERVAL_MS'], variableHelp: { - JENKINS_BASE_URL: - variableHelp('Jenkins 服务端 API 地址,只能由后端服务访问。', { + JENKINS_BASE_URL: variableHelp( + 'Jenkins 服务端 API 地址,只能由后端服务访问。', + { docUrl: 'https://www.jenkins.io/doc/book/using/remote-access-api/', - }), - JENKINS_USERNAME: variableHelp('用于调用 Jenkins Remote API 的服务账号。'), + }, + ), + JENKINS_USERNAME: variableHelp( + '用于调用 Jenkins Remote API 的服务账号。', + ), JENKINS_API_TOKEN: variableHelp( 'Jenkins 服务账号 API token,只能保存在服务端环境变量或后续加密密钥表。', ), JENKINS_AUTO_SYNC_ENABLED: variableHelp( - '是否启用轻量自动同步器,自动扫描 queued/running 发布单。', + '是否启用 Jenkins 自动同步器;配置 REDIS_URL 时使用 BullMQ 调度,否则降级进程内定时器。', { example: 'true' }, ), JENKINS_AUTO_SYNC_INTERVAL_MS: variableHelp( @@ -57,7 +64,9 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [ optional: ['GITEA_WEBHOOK_SECRET', 'GITEA_WEBHOOK_RELAY_BASE_URL'], note: 'GITEA_WEBHOOK_SECRET 用于开启 webhook 签名校验;GITEA_WEBHOOK_RELAY_BASE_URL 用于校验仓库 push hook 是否已接入 Jenkins relay。', variableHelp: { - GITEA_BASE_URL: variableHelp('Gitea 服务端 API 地址,用于读取仓库、分支和 tag。'), + GITEA_BASE_URL: variableHelp( + 'Gitea 服务端 API 地址,用于读取仓库、分支和 tag。', + ), GITEA_TOKEN: variableHelp( 'Gitea API token,只能由后端持有,用于读取 refs 和后续仓库元数据。', { docUrl: 'https://docs.gitea.com/development/api-usage' }, @@ -108,9 +117,12 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [ NOTIFICATION_WEBHOOK_URL: variableHelp( '内部通知网关或通用 webhook 的服务端地址,适合接自建消息中转服务。', ), - DEVOPS_PUBLIC_URL: variableHelp('通知消息中跳转回 DevOps 平台的公开访问地址。', { - example: 'https://devops.mrzhan.top', - }), + DEVOPS_PUBLIC_URL: variableHelp( + '通知消息中跳转回 DevOps 平台的公开访问地址。', + { + example: 'https://devops.mrzhan.top', + }, + ), NOTIFICATION_OUTBOX_RETRY_ENABLED: variableHelp( '是否启用轻量通知 outbox 重试器,失败通知会按 nextAttemptAt 自动重投。', { example: 'true' }, @@ -236,7 +248,9 @@ export class SettingsService { private toNotificationIntegrationStatus( definition: IntegrationDefinition, ): IntegrationConfigStatus { - const provider = this.getConfigValue('NOTIFICATION_PROVIDER')?.toLowerCase(); + const provider = this.getConfigValue( + 'NOTIFICATION_PROVIDER', + )?.toLowerCase(); const providerWebhook = this.notificationWebhookKey(provider); const required: EnvStringKey[] = providerWebhook ? ['NOTIFICATION_PROVIDER', providerWebhook]