From 0d60820c938600ee1e5f7cdfb0c96a4244028c9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B9=9B=E5=85=AE?= Date: Fri, 12 Jun 2026 03:39:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=E9=80=9A=E7=9F=A5=20?= =?UTF-8?q?outbox=20=E9=87=8D=E8=AF=95=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - notifications: 支持失败 outbox 按配置 nextAttemptAt 重试并达到上限转 dead - settings: 暴露通知 outbox 重试相关环境变量说明 --- src/config/env.schema.ts | 5 + .../outbox/notification-outbox.service.ts | 50 +++- .../outbox/notification-outbox.types.ts | 8 +- .../wecom/deploy-notification.service.ts | 242 +++++++++++++++++- .../notification-outbox-retry.service.ts | 114 +++++++++ src/notifications/wecom/wecom.module.ts | 7 +- src/settings/settings.service.ts | 25 ++ 7 files changed, 446 insertions(+), 5 deletions(-) create mode 100644 src/notifications/wecom/notification-outbox-retry.service.ts diff --git a/src/config/env.schema.ts b/src/config/env.schema.ts index 91d1107..d618b7f 100644 --- a/src/config/env.schema.ts +++ b/src/config/env.schema.ts @@ -70,6 +70,11 @@ export const envSchema = z.object({ NOTIFICATION_WEBHOOK_URL: optionalUrl, FEISHU_WEBHOOK_URL: optionalUrl, WECOM_WEBHOOK_URL: optionalUrl, + NOTIFICATION_OUTBOX_RETRY_ENABLED: booleanFromEnv, + NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS: positiveIntegerFromEnv(30000), + NOTIFICATION_OUTBOX_RETRY_DELAY_MS: positiveIntegerFromEnv(60000), + NOTIFICATION_OUTBOX_MAX_ATTEMPTS: positiveIntegerFromEnv(3), + NOTIFICATION_OUTBOX_BATCH_SIZE: positiveIntegerFromEnv(10), LLM_BASE_URL: optionalUrl, LLM_API_KEY: optionalString, LLM_MODEL: optionalString, diff --git a/src/notifications/outbox/notification-outbox.service.ts b/src/notifications/outbox/notification-outbox.service.ts index 785dc63..ef2d6db 100644 --- a/src/notifications/outbox/notification-outbox.service.ts +++ b/src/notifications/outbox/notification-outbox.service.ts @@ -10,6 +10,7 @@ import { PrismaService } from '../../prisma/prisma.service'; import { CompleteNotificationOutboxInput, CreateNotificationOutboxInput, + ListRetryableNotificationOutboxInput, NotificationOutboxMessage, NotificationOutboxStatus, } from './notification-outbox.types'; @@ -20,6 +21,7 @@ const statusToPrisma: Record< > = { sent: OutboxStatus.SENT, failed: OutboxStatus.FAILED, + dead: OutboxStatus.DEAD, }; const statusFromPrisma: Record = { @@ -120,8 +122,8 @@ export class NotificationOutboxService { attemptCount: { increment: 1, }, - lastError: input.lastError, - nextAttemptAt: input.nextAttemptAt, + lastError: input.lastError ?? null, + nextAttemptAt: input.nextAttemptAt ?? null, }, }); @@ -145,6 +147,50 @@ export class NotificationOutboxService { return messages.map((message) => this.toSummary(message)); } + async listRetryableMessages( + input: ListRetryableNotificationOutboxInput, + ): Promise { + const prisma = this.databaseClient(); + + if (!prisma) { + return this.memoryMessages + .filter((message) => { + if (message.status !== 'failed') { + return false; + } + + if (message.attemptCount >= input.maxAttempts) { + return false; + } + + if (!message.nextAttemptAt) { + return false; + } + + return new Date(message.nextAttemptAt).getTime() <= input.now.getTime(); + }) + .slice(0, input.take); + } + + const messages = await prisma.outboxMessage.findMany({ + where: { + status: OutboxStatus.FAILED, + attemptCount: { + lt: input.maxAttempts, + }, + nextAttemptAt: { + lte: input.now, + }, + }, + orderBy: { + nextAttemptAt: 'asc', + }, + take: input.take, + }); + + return messages.map((message) => this.toSummary(message)); + } + private toSummary(message: OutboxMessage): NotificationOutboxMessage { return { id: message.id, diff --git a/src/notifications/outbox/notification-outbox.types.ts b/src/notifications/outbox/notification-outbox.types.ts index bddc305..c8a79b4 100644 --- a/src/notifications/outbox/notification-outbox.types.ts +++ b/src/notifications/outbox/notification-outbox.types.ts @@ -27,7 +27,13 @@ export type CreateNotificationOutboxInput = { }; export type CompleteNotificationOutboxInput = { - status: 'sent' | 'failed'; + status: 'sent' | 'failed' | 'dead'; lastError?: string; nextAttemptAt?: Date; }; + +export type ListRetryableNotificationOutboxInput = { + now: Date; + maxAttempts: number; + take: number; +}; diff --git a/src/notifications/wecom/deploy-notification.service.ts b/src/notifications/wecom/deploy-notification.service.ts index b1e94e0..6c9a01e 100644 --- a/src/notifications/wecom/deploy-notification.service.ts +++ b/src/notifications/wecom/deploy-notification.service.ts @@ -33,6 +33,21 @@ type DeployNotificationHealth = { message?: string; }; +type DeployNotificationPayloadSummary = { + event?: string; + title?: string; + deployRunId?: string; + projectKey?: string; + environment?: string; + ref?: string; + status?: string; + operator?: string; + jenkinsQueueId?: string | number; + jenkinsBuildNumber?: number; + failureSummary?: string; + publicUrl?: string; +}; + /** * 统一封装发布单通知，保证 webhook 只存在于后端配置边界内。 */ @@ -138,6 +153,68 @@ export class DeployNotificationService { } } + async retryOutboxMessage( + message: NotificationOutboxMessage, + ): Promise { + const health = this.healthSummary(); + const channel = health.provider ?? 'none'; + + if (health.status === 'not_configured') { + await this.completeOutboxFailure( + message, + health.message ?? '通知平台尚未配置', + ); + + return { + status: 'skipped', + channel, + reason: health.message ?? '通知平台尚未配置', + outboxMessageId: message.id, + }; + } + + if (health.status === 'unavailable' || !health.provider) { + await this.completeOutboxFailure( + message, + health.message ?? '通知平台不可用', + ); + + return { + status: 'failed', + channel, + reason: health.message ?? '通知平台不可用', + outboxMessageId: message.id, + }; + } + + try { + await this.sendOutboxPayloadByProvider(health.provider, message); + await this.notificationOutboxService.completeMessage(message.id, { + status: 'sent', + }); + + return { + status: 'sent', + channel: health.provider, + outboxMessageId: message.id, + }; + } catch (error) { + const reason = + error instanceof Error + ? redactSensitive({ message: error.message }).message + : '未知通知错误'; + + await this.completeOutboxFailure(message, reason); + + return { + status: 'failed', + channel: health.provider, + reason, + outboxMessageId: message.id, + }; + } + } + private async createOutboxMessage( run: DeployRunSummary, event: DeployNotificationEvent, @@ -156,9 +233,25 @@ export class DeployNotificationService { message: NotificationOutboxMessage, reason: string, ): Promise { + const nextAttemptCount = message.attemptCount + 1; + const maxAttempts = this.config.get('NOTIFICATION_OUTBOX_MAX_ATTEMPTS', { + infer: true, + }); + const retryEnabled = this.config.get('NOTIFICATION_OUTBOX_RETRY_ENABLED', { + infer: true, + }); + const retryDelayMs = this.config.get('NOTIFICATION_OUTBOX_RETRY_DELAY_MS', { + infer: true, + }); + const exhausted = nextAttemptCount >= maxAttempts; + await this.notificationOutboxService.completeMessage(message.id, { - status: 'failed', + status: exhausted ? 'dead' : 'failed', lastError: reason, + nextAttemptAt: + retryEnabled && !exhausted + ? new Date(Date.now() + retryDelayMs) + : undefined, }); } @@ -239,6 +332,62 @@ export class DeployNotificationService { ); } + private async sendOutboxPayloadByProvider( + provider: DeployNotificationProvider, + message: NotificationOutboxMessage, + ): Promise { + const payload = this.toNotificationPayloadSummary(message.payloadSummary); + const title = payload.title || this.titleFor(this.eventFromTemplate(message)); + const markdown = this.markdownForPayload(payload, title); + const plainText = this.plainTextForPayload(payload, title); + + if (provider === 'wecom') { + await this.weComClient.sendMarkdown({ + title, + markdown, + }); + return; + } + + if (provider === 'feishu') { + await this.postWebhook( + this.config.get('FEISHU_WEBHOOK_URL', { infer: true }) ?? '', + { + msg_type: 'text', + content: { + text: `${title}\n${plainText}`, + }, + }, + provider, + ); + return; + } + + await this.postWebhook( + this.config.get('NOTIFICATION_WEBHOOK_URL', { infer: true }) ?? '', + { + event: payload.event ?? message.template, + title, + markdown, + run: { + id: payload.deployRunId ?? message.deployRunId, + projectKey: payload.projectKey, + environment: payload.environment, + ref: payload.ref, + status: payload.status, + operator: payload.operator, + jenkinsQueueId: payload.jenkinsQueueId, + jenkinsBuildNumber: payload.jenkinsBuildNumber, + failureSummary: payload.failureSummary, + }, + publicUrl: + payload.publicUrl ?? + this.config.get('DEVOPS_PUBLIC_URL', { infer: true }), + }, + provider, + ); + } + private async postWebhook( webhookUrl: string, body: unknown, @@ -358,4 +507,95 @@ export class DeployNotificationService { .filter(Boolean) .join('\n'); } + + private eventFromTemplate( + message: NotificationOutboxMessage, + ): DeployNotificationEvent { + const event = message.template.replace(/^deploy\./, ''); + + if ( + event === 'created' || + event === 'queued' || + event === 'succeeded' || + event === 'failed' || + event === 'canceled' || + event === 'retried' + ) { + return event; + } + + return 'failed'; + } + + private toNotificationPayloadSummary( + value: Record | undefined, + ): DeployNotificationPayloadSummary { + return { + event: this.optionalString(value?.event), + title: this.optionalString(value?.title), + deployRunId: this.optionalString(value?.deployRunId), + projectKey: this.optionalString(value?.projectKey), + environment: this.optionalString(value?.environment), + ref: this.optionalString(value?.ref), + status: this.optionalString(value?.status), + operator: this.optionalString(value?.operator), + jenkinsQueueId: + typeof value?.jenkinsQueueId === 'string' || + typeof value?.jenkinsQueueId === 'number' + ? value.jenkinsQueueId + : undefined, + jenkinsBuildNumber: + typeof value?.jenkinsBuildNumber === 'number' + ? value.jenkinsBuildNumber + : undefined, + failureSummary: this.optionalString(value?.failureSummary), + publicUrl: this.optionalString(value?.publicUrl), + }; + } + + private markdownForPayload( + payload: DeployNotificationPayloadSummary, + title: string, + ): string { + return [ + `> 事件：${title}`, + payload.projectKey ? `> 项目：${payload.projectKey}` : undefined, + payload.environment ? `> 环境：${payload.environment}` : undefined, + payload.ref ? `> Ref：${payload.ref}` : undefined, + payload.status ? `> 状态：${payload.status}` : undefined, + payload.operator ? `> 操作者：${payload.operator}` : undefined, + payload.deployRunId ? `> Run ID：${payload.deployRunId}` : undefined, + payload.jenkinsQueueId + ? `> Jenkins Queue：${payload.jenkinsQueueId}` + : undefined, + payload.jenkinsBuildNumber + ? `> Jenkins Build：#${payload.jenkinsBuildNumber}` + : undefined, + payload.failureSummary ? `> 失败摘要：${payload.failureSummary}` : undefined, + ] + .filter(Boolean) + .join('\n'); + } + + private plainTextForPayload( + payload: DeployNotificationPayloadSummary, + title: string, + ): string { + return [ + `事件：${title}`, + payload.projectKey ? `项目：${payload.projectKey}` : undefined, + payload.environment ? `环境：${payload.environment}` : undefined, + payload.ref ? `Ref：${payload.ref}` : undefined, + payload.status ? `状态：${payload.status}` : undefined, + payload.operator ? `操作者：${payload.operator}` : undefined, + payload.deployRunId ? `Run ID：${payload.deployRunId}` : undefined, + payload.failureSummary ? `失败摘要：${payload.failureSummary}` : undefined, + ] + .filter(Boolean) + .join('\n'); + } + + private optionalString(value: unknown): string | undefined { + return typeof value === 'string' && value.trim() ? value : undefined; + } } diff --git a/src/notifications/wecom/notification-outbox-retry.service.ts b/src/notifications/wecom/notification-outbox-retry.service.ts new file mode 100644 index 0000000..034f3b3 --- /dev/null +++ b/src/notifications/wecom/notification-outbox-retry.service.ts @@ -0,0 +1,114 @@ +/** + * 通知 outbox 重试器复用当前 provider 投递边界，后续可迁移到 BullMQ worker。 + */ +import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { AuditService } from '../../audit/audit.service'; +import { redactSensitive } from '../../common/security/redact-sensitive'; +import { EnvConfig } from '../../config/env.schema'; +import { NotificationOutboxService } from '../outbox/notification-outbox.service'; +import { DeployNotificationService } from './deploy-notification.service'; + +@Injectable() +export class NotificationOutboxRetryService + implements OnModuleInit, OnModuleDestroy +{ + private timer: ReturnType | undefined; + private running = false; + + constructor( + private readonly config: ConfigService, + private readonly notificationOutboxService: NotificationOutboxService, + private readonly deployNotificationService: DeployNotificationService, + private readonly auditService: AuditService, + ) {} + + onModuleInit(): void { + if ( + !this.config.get('NOTIFICATION_OUTBOX_RETRY_ENABLED', { infer: true }) + ) { + return; + } + + const intervalMs = this.config.get( + 'NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS', + { infer: true }, + ); + + this.timer = setInterval(() => { + void this.runOnce(); + }, intervalMs); + void this.runOnce(); + } + + onModuleDestroy(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = undefined; + } + } + + async runOnce(): Promise { + if (this.running) { + return; + } + + this.running = true; + + try { + const maxAttempts = this.config.get('NOTIFICATION_OUTBOX_MAX_ATTEMPTS', { + infer: true, + }); + const batchSize = this.config.get('NOTIFICATION_OUTBOX_BATCH_SIZE', { + infer: true, + }); + const messages = await this.notificationOutboxService.listRetryableMessages( + { + now: new Date(), + maxAttempts, + take: batchSize, + }, + ); + let sentCount = 0; + let failedCount = 0; + let skippedCount = 0; + + for (const message of messages) { + const result = + await this.deployNotificationService.retryOutboxMessage(message); + + if (result.status === 'sent') { + sentCount += 1; + } else if (result.status === 'skipped') { + skippedCount += 1; + } else { + failedCount += 1; + } + } + + await this.auditService.record({ + action: 'NOTIFICATION_OUTBOX_RETRY_TICK_COMPLETED', + resourceType: 'notification_outbox', + after: { + totalCandidates: messages.length, + sentCount, + skippedCount, + failedCount, + }, + }); + } catch (error) { + const message = + error instanceof Error ? error.message : '通知 outbox 自动重试失败'; + + await this.auditService.record({ + action: 'NOTIFICATION_OUTBOX_RETRY_TICK_FAILED', + resourceType: 'notification_outbox', + after: { + message: redactSensitive({ message }).message, + }, + }); + } finally { + this.running = false; + } + } +} diff --git a/src/notifications/wecom/wecom.module.ts b/src/notifications/wecom/wecom.module.ts index 1b58161..bbef3f2 100644 --- a/src/notifications/wecom/wecom.module.ts +++ b/src/notifications/wecom/wecom.module.ts @@ -4,11 +4,16 @@ import { Module } from '@nestjs/common'; import { NotificationOutboxModule } from '../outbox/notification-outbox.module'; import { DeployNotificationService } from './deploy-notification.service'; +import { NotificationOutboxRetryService } from './notification-outbox-retry.service'; import { WeComClient } from './wecom.client'; @Module({ imports: [NotificationOutboxModule], - providers: [WeComClient, DeployNotificationService], + providers: [ + WeComClient, + DeployNotificationService, + NotificationOutboxRetryService, + ], exports: [WeComClient, DeployNotificationService], }) export class WeComModule {} diff --git a/src/settings/settings.service.ts b/src/settings/settings.service.ts index d859f60..6f6b0b3 100644 --- a/src/settings/settings.service.ts +++ b/src/settings/settings.service.ts @@ -76,6 +76,11 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [ 'FEISHU_WEBHOOK_URL', 'NOTIFICATION_WEBHOOK_URL', 'DEVOPS_PUBLIC_URL', + 'NOTIFICATION_OUTBOX_RETRY_ENABLED', + 'NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS', + 'NOTIFICATION_OUTBOX_RETRY_DELAY_MS', + 'NOTIFICATION_OUTBOX_MAX_ATTEMPTS', + 'NOTIFICATION_OUTBOX_BATCH_SIZE', ], note: '当前不强制配置机器人；后续可按 provider 接入企微、飞书或通用 webhook。', variableHelp: { @@ -102,6 +107,26 @@ const INTEGRATION_DEFINITIONS: IntegrationDefinition[] = [ DEVOPS_PUBLIC_URL: variableHelp('通知消息中跳转回 DevOps 平台的公开访问地址。', { example: 'https://devops.mrzhan.top', }), + NOTIFICATION_OUTBOX_RETRY_ENABLED: variableHelp( + '是否启用轻量通知 outbox 重试器，失败通知会按 nextAttemptAt 自动重投。', + { example: 'true' }, + ), + NOTIFICATION_OUTBOX_RETRY_INTERVAL_MS: variableHelp( + '通知 outbox 重试扫描间隔，单位毫秒。', + { example: '30000' }, + ), + NOTIFICATION_OUTBOX_RETRY_DELAY_MS: variableHelp( + '单条通知投递失败后的下次重试延迟，单位毫秒。', + { example: '60000' }, + ), + NOTIFICATION_OUTBOX_MAX_ATTEMPTS: variableHelp( + '单条通知最大投递次数，达到后标记为 dead，避免无限重试。', + { example: '3' }, + ), + NOTIFICATION_OUTBOX_BATCH_SIZE: variableHelp( + '每轮 outbox 重试最多处理的消息数量。', + { example: '10' }, + ), }, }, {