From 4597dcaa3864da2473ebf4183fcf3e1b7d5fa874 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B9=9B=E5=85=AE?= Date: Fri, 12 Jun 2026 03:22:38 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8C=81=E4=B9=85=E5=8C=96=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=20outbox=20=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - notifications: 通知发送前写入 outbox，并在成功、失败或未配置时回写状态 - deploy-runs: 审计通知结果时带上 outboxMessageId - outbox: 新增超级管理员查询最近通知投递摘要接口 --- src/deploy-runs/deploy-execution.service.ts | 2 + src/deploy-runs/deploy-runs.service.ts | 4 + .../outbox/notification-outbox.controller.ts | 27 +++ .../outbox/notification-outbox.module.ts | 13 ++ .../outbox/notification-outbox.service.ts | 187 ++++++++++++++++++ .../outbox/notification-outbox.types.ts | 33 ++++ .../wecom/deploy-notification.service.spec.ts | 34 +++- .../wecom/deploy-notification.service.ts | 111 ++++++++++- src/notifications/wecom/wecom.module.ts | 5 + 9 files changed, 406 insertions(+), 10 deletions(-) create mode 100644 src/notifications/outbox/notification-outbox.controller.ts create mode 100644 src/notifications/outbox/notification-outbox.module.ts create mode 100644 src/notifications/outbox/notification-outbox.service.ts create mode 100644 src/notifications/outbox/notification-outbox.types.ts diff --git a/src/deploy-runs/deploy-execution.service.ts b/src/deploy-runs/deploy-execution.service.ts index b6120b9..4549c21 100644 --- a/src/deploy-runs/deploy-execution.service.ts +++ b/src/deploy-runs/deploy-execution.service.ts @@ -361,6 +361,7 @@ export class DeployExecutionService { event, channel: notification.channel, reason: notification.reason, + outboxMessageId: notification.outboxMessageId, }, }); @@ -547,6 +548,7 @@ export class DeployExecutionService { event, channel: notification.channel, reason: notification.reason, + outboxMessageId: notification.outboxMessageId, }, }); } diff --git a/src/deploy-runs/deploy-runs.service.ts b/src/deploy-runs/deploy-runs.service.ts index 5fe8dfb..8b5d707 100644 --- a/src/deploy-runs/deploy-runs.service.ts +++ b/src/deploy-runs/deploy-runs.service.ts @@ -1,3 +1,6 @@ +/** + * 发布记录应用服务负责创建、取消、同步发布单，并归档审计与通知结果。 + */ import { Injectable } from '@nestjs/common'; import { AuthService } from '../auth/auth.service'; import { AuthenticatedUser } from '../auth/auth.types'; @@ -370,6 +373,7 @@ export class DeployRunsService { event, channel: notification.channel, reason: notification.reason, + outboxMessageId: notification.outboxMessageId, }, }); } diff --git a/src/notifications/outbox/notification-outbox.controller.ts b/src/notifications/outbox/notification-outbox.controller.ts new file mode 100644 index 0000000..36c5631 --- /dev/null +++ b/src/notifications/outbox/notification-outbox.controller.ts @@ -0,0 +1,27 @@ +/** + * 通知 outbox 查询入口只允许超级管理员查看脱敏投递摘要。 + */ +import { Controller, Get, UseGuards } from '@nestjs/common'; +import { ApiOkResponse, ApiTags } from '@nestjs/swagger'; +import { AdminOnlyMessage } from '../../auth/admin-only-message.decorator'; +import { AuthGuard } from '../../auth/auth.guard'; +import { PasswordChangeGuard } from '../../auth/password-change.guard'; +import { SuperAdminGuard } from '../../auth/super-admin.guard'; +import { NotificationOutboxService } from './notification-outbox.service'; +import { NotificationOutboxMessage } from './notification-outbox.types'; + +@ApiTags('通知 outbox') +@Controller('notification-outbox') +@UseGuards(AuthGuard, PasswordChangeGuard, SuperAdminGuard) +@AdminOnlyMessage('无权限访问通知 outbox') +export class NotificationOutboxController { + constructor( + private readonly notificationOutboxService: NotificationOutboxService, + ) {} + + @Get() + @ApiOkResponse({ description: '查询最近的通知 outbox 投递摘要。' }) + async listRecentMessages(): Promise { + return this.notificationOutboxService.listRecentMessages(); + } +} diff --git a/src/notifications/outbox/notification-outbox.module.ts b/src/notifications/outbox/notification-outbox.module.ts new file mode 100644 index 0000000..96b6fef --- /dev/null +++ b/src/notifications/outbox/notification-outbox.module.ts @@ -0,0 +1,13 @@ +/** + * 通知 outbox 模块提供投递记录写入服务和超级管理员查询入口。 + */ +import { Module } from '@nestjs/common'; +import { NotificationOutboxController } from './notification-outbox.controller'; +import { NotificationOutboxService } from './notification-outbox.service'; + +@Module({ + controllers: [NotificationOutboxController], + providers: [NotificationOutboxService], + exports: [NotificationOutboxService], +}) +export class NotificationOutboxModule {} diff --git a/src/notifications/outbox/notification-outbox.service.ts b/src/notifications/outbox/notification-outbox.service.ts new file mode 100644 index 0000000..785dc63 --- /dev/null +++ b/src/notifications/outbox/notification-outbox.service.ts @@ -0,0 +1,187 @@ +/** + * 通知 outbox 负责在真正投递前持久化通知事件，并记录后续投递结果。 + */ +import { Injectable } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { OutboxMessage, OutboxStatus, Prisma } from '@prisma/client'; +import { redactSensitive } from '../../common/security/redact-sensitive'; +import { EnvConfig } from '../../config/env.schema'; +import { PrismaService } from '../../prisma/prisma.service'; +import { + CompleteNotificationOutboxInput, + CreateNotificationOutboxInput, + NotificationOutboxMessage, + NotificationOutboxStatus, +} from './notification-outbox.types'; + +const statusToPrisma: Record< + CompleteNotificationOutboxInput['status'], + OutboxStatus +> = { + sent: OutboxStatus.SENT, + failed: OutboxStatus.FAILED, +}; + +const statusFromPrisma: Record = { + PENDING: 'pending', + SENT: 'sent', + FAILED: 'failed', + DEAD: 'dead', +}; + +@Injectable() +export class NotificationOutboxService { + private readonly memoryMessages: NotificationOutboxMessage[] = []; + + constructor( + private readonly config: ConfigService, + private readonly prisma: PrismaService, + ) {} + + async createMessage( + input: CreateNotificationOutboxInput, + ): Promise { + const payload = this.toJsonValue(redactSensitive(input.payload)); + const prisma = this.databaseClient(); + + if (!prisma) { + const existing = this.memoryMessages.find( + (message) => message.idempotencyKey === input.idempotencyKey, + ); + + if (existing) { + return existing; + } + + const now = new Date().toISOString(); + const message: NotificationOutboxMessage = { + id: `outbox_${Date.now()}_${this.memoryMessages.length + 1}`, + deployRunId: input.deployRunId, + channel: input.channel, + template: input.template, + payloadSummary: this.jsonToRecord(payload), + status: 'pending', + attemptCount: 0, + idempotencyKey: input.idempotencyKey, + createdAt: now, + updatedAt: now, + }; + + this.memoryMessages.unshift(message); + return message; + } + + const message = await prisma.outboxMessage.upsert({ + where: { + idempotencyKey: input.idempotencyKey, + }, + update: { + channel: input.channel, + template: input.template, + payload, + }, + create: { + deployRunId: input.deployRunId, + channel: input.channel, + template: input.template, + payload, + idempotencyKey: input.idempotencyKey, + }, + }); + + return this.toSummary(message); + } + + async completeMessage( + id: string, + input: CompleteNotificationOutboxInput, + ): Promise { + const prisma = this.databaseClient(); + + if (!prisma) { + const message = this.memoryMessages.find((item) => item.id === id); + + if (!message) { + return undefined; + } + + message.status = input.status; + message.attemptCount += 1; + message.lastError = input.lastError; + message.nextAttemptAt = input.nextAttemptAt?.toISOString(); + message.updatedAt = new Date().toISOString(); + return message; + } + + const message = await prisma.outboxMessage.update({ + where: { id }, + data: { + status: statusToPrisma[input.status], + attemptCount: { + increment: 1, + }, + lastError: input.lastError, + nextAttemptAt: input.nextAttemptAt, + }, + }); + + return this.toSummary(message); + } + + async listRecentMessages(): Promise { + const prisma = this.databaseClient(); + + if (!prisma) { + return [...this.memoryMessages].slice(0, 50); + } + + const messages = await prisma.outboxMessage.findMany({ + orderBy: { + createdAt: 'desc', + }, + take: 50, + }); + + return messages.map((message) => this.toSummary(message)); + } + + private toSummary(message: OutboxMessage): NotificationOutboxMessage { + return { + id: message.id, + deployRunId: message.deployRunId ?? undefined, + channel: message.channel, + template: message.template, + payloadSummary: this.jsonToRecord(message.payload), + status: statusFromPrisma[message.status], + attemptCount: message.attemptCount, + nextAttemptAt: message.nextAttemptAt?.toISOString(), + lastError: message.lastError ?? undefined, + idempotencyKey: message.idempotencyKey, + createdAt: message.createdAt.toISOString(), + updatedAt: message.updatedAt.toISOString(), + }; + } + + private toJsonValue(value: unknown): Prisma.InputJsonValue { + return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue; + } + + private jsonToRecord(value: unknown): Record | undefined { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return undefined; + } + + return value as Record; + } + + private databaseClient(): PrismaService | undefined { + if ( + !this.config.get('USE_DATABASE_READS', { infer: true }) || + !this.config.get('DATABASE_URL', { infer: true }) + ) { + return undefined; + } + + return this.prisma; + } +} diff --git a/src/notifications/outbox/notification-outbox.types.ts b/src/notifications/outbox/notification-outbox.types.ts new file mode 100644 index 0000000..bddc305 --- /dev/null +++ b/src/notifications/outbox/notification-outbox.types.ts @@ -0,0 +1,33 @@ +/** + * 通知 outbox 只暴露脱敏后的投递摘要，供运维页面确认通知链路状态。 + */ +export type NotificationOutboxStatus = 'pending' | 'sent' | 'failed' | 'dead'; + +export type NotificationOutboxMessage = { + id: string; + deployRunId?: string; + channel: string; + template: string; + payloadSummary?: Record; + status: NotificationOutboxStatus; + attemptCount: number; + nextAttemptAt?: string; + lastError?: string; + idempotencyKey: string; + createdAt: string; + updatedAt: string; +}; + +export type CreateNotificationOutboxInput = { + deployRunId?: string; + channel: string; + template: string; + payload: Record; + idempotencyKey: string; +}; + +export type CompleteNotificationOutboxInput = { + status: 'sent' | 'failed'; + lastError?: string; + nextAttemptAt?: Date; +}; diff --git a/src/notifications/wecom/deploy-notification.service.spec.ts b/src/notifications/wecom/deploy-notification.service.spec.ts index 819af68..31de650 100644 --- a/src/notifications/wecom/deploy-notification.service.spec.ts +++ b/src/notifications/wecom/deploy-notification.service.spec.ts @@ -1,12 +1,13 @@ +/** + * 覆盖通知 provider 与 outbox 组合边界：缺配置时只记录摘要，配置后才投递。 + */ import { ConfigService } from '@nestjs/config'; import { EnvConfig } from '../../config/env.schema'; import { WeComClient } from './wecom.client'; import { DeployNotificationService } from './deploy-notification.service'; import { DeployRunSummary } from '../../deploy-runs/deploy-run.types'; +import { NotificationOutboxService } from '../outbox/notification-outbox.service'; -/** - * 覆盖通知 provider 的安全边界：缺少配置时跳过，配置后才发送。 - */ describe('DeployNotificationService', () => { const run: DeployRunSummary = { id: 'run_001', @@ -35,6 +36,7 @@ describe('DeployNotificationService', () => { const service = new DeployNotificationService( createConfig({}), weCom as unknown as WeComClient, + createOutbox() as unknown as NotificationOutboxService, ); const result = await service.sendDeployRunEvent(run, 'created'); @@ -52,6 +54,7 @@ describe('DeployNotificationService', () => { const service = new DeployNotificationService( createConfig({ NOTIFICATION_PROVIDER: 'wecom' }), weCom as unknown as WeComClient, + createOutbox() as unknown as NotificationOutboxService, ); const result = await service.sendDeployRunEvent(run, 'created'); @@ -80,6 +83,7 @@ describe('DeployNotificationService', () => { FEISHU_WEBHOOK_URL: 'https://open.feishu.cn/open-apis/bot/v2/hook/test', }), weCom as unknown as WeComClient, + createOutbox() as unknown as NotificationOutboxService, ); const result = await service.sendDeployRunEvent(run, 'failed'); @@ -109,6 +113,7 @@ describe('DeployNotificationService', () => { DEVOPS_PUBLIC_URL: 'https://devops.mrzhan.top', }), weCom as unknown as WeComClient, + createOutbox() as unknown as NotificationOutboxService, ); const result = await service.sendDeployRunEvent(run, 'succeeded'); @@ -131,3 +136,26 @@ function createConfig( get: jest.fn((key: keyof EnvConfig) => values[key]), } as unknown as ConfigService; } + +function createOutbox(): Pick< + NotificationOutboxService, + 'createMessage' | 'completeMessage' +> { + return { + createMessage: jest.fn((input) => + Promise.resolve({ + id: 'outbox_001', + deployRunId: input.deployRunId, + channel: input.channel, + template: input.template, + payloadSummary: input.payload, + status: 'pending' as const, + attemptCount: 0, + idempotencyKey: input.idempotencyKey, + createdAt: '2026-06-11T00:00:00.000Z', + updatedAt: '2026-06-11T00:00:00.000Z', + }), + ), + completeMessage: jest.fn(), + }; +} diff --git a/src/notifications/wecom/deploy-notification.service.ts b/src/notifications/wecom/deploy-notification.service.ts index 44d46d8..b1e94e0 100644 --- a/src/notifications/wecom/deploy-notification.service.ts +++ b/src/notifications/wecom/deploy-notification.service.ts @@ -1,8 +1,13 @@ +/** + * 发布通知适配服务封装 provider 选择、outbox 记录和后端 webhook 投递。 + */ import { Injectable } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { redactSensitive } from '../../common/security/redact-sensitive'; import { EnvConfig } from '../../config/env.schema'; import { DeployRunSummary } from '../../deploy-runs/deploy-run.types'; +import { NotificationOutboxService } from '../outbox/notification-outbox.service'; +import { NotificationOutboxMessage } from '../outbox/notification-outbox.types'; import { WeComClient } from './wecom.client'; export type DeployNotificationEvent = @@ -17,6 +22,7 @@ export type DeployNotificationResult = { status: 'sent' | 'skipped' | 'failed'; channel: DeployNotificationProvider | 'none'; reason?: string; + outboxMessageId?: string; }; export type DeployNotificationProvider = 'wecom' | 'feishu' | 'generic'; @@ -35,6 +41,7 @@ export class DeployNotificationService { constructor( private readonly config: ConfigService, private readonly weComClient: WeComClient, + private readonly notificationOutboxService: NotificationOutboxService, ) {} healthSummary(): DeployNotificationHealth { @@ -72,39 +79,89 @@ export class DeployNotificationService { event: DeployNotificationEvent, ): Promise { const health = this.healthSummary(); + const channel = health.provider ?? 'none'; + const outboxMessage = await this.createOutboxMessage(run, event, channel); if (health.status === 'not_configured') { + await this.completeOutboxFailure( + outboxMessage, + health.message ?? '通知平台尚未配置', + ); + return { status: 'skipped', - channel: health.provider ?? 'none', + channel, reason: health.message ?? '通知平台尚未配置', + outboxMessageId: outboxMessage.id, }; } if (health.status === 'unavailable' || !health.provider) { + await this.completeOutboxFailure( + outboxMessage, + health.message ?? '通知平台不可用', + ); + return { status: 'failed', - channel: health.provider ?? 'none', + channel, reason: health.message ?? '通知平台不可用', + outboxMessageId: outboxMessage.id, }; } try { await this.sendByProvider(health.provider, run, event); + await this.notificationOutboxService.completeMessage(outboxMessage.id, { + status: 'sent', + }); - return { status: 'sent', channel: health.provider }; + return { + status: 'sent', + channel: health.provider, + outboxMessageId: outboxMessage.id, + }; } catch (error) { + const reason = + error instanceof Error + ? redactSensitive({ message: error.message }).message + : '未知通知错误'; + + await this.completeOutboxFailure(outboxMessage, reason); + return { status: 'failed', channel: health.provider, - reason: - error instanceof Error - ? redactSensitive({ message: error.message }).message - : '未知通知错误', + reason, + outboxMessageId: outboxMessage.id, }; } } + private async createOutboxMessage( + run: DeployRunSummary, + event: DeployNotificationEvent, + channel: DeployNotificationProvider | 'none', + ): Promise { + return this.notificationOutboxService.createMessage({ + deployRunId: run.id, + channel, + template: `deploy.${event}`, + payload: this.outboxPayloadFor(run, event), + idempotencyKey: this.outboxIdempotencyKey(run, event), + }); + } + + private async completeOutboxFailure( + message: NotificationOutboxMessage, + reason: string, + ): Promise { + await this.notificationOutboxService.completeMessage(message.id, { + status: 'failed', + lastError: reason, + }); + } + private resolveProvider(): DeployNotificationProvider | undefined { const provider = this.config .get('NOTIFICATION_PROVIDER', { infer: true }) @@ -224,6 +281,46 @@ export class DeployNotificationService { return 'DevOps 发布已创建'; } + private outboxPayloadFor( + run: DeployRunSummary, + event: DeployNotificationEvent, + ): Record { + return { + event: `deploy.${event}`, + title: this.titleFor(event), + deployRunId: run.id, + projectKey: run.projectKey, + environment: run.environment, + ref: this.safeRefForOutbox(run.ref), + status: run.status, + operator: run.operator, + jenkinsQueueId: run.jenkinsQueueId, + jenkinsBuildNumber: run.jenkinsBuildNumber, + failureSummary: run.failureSummary, + publicUrl: this.config.get('DEVOPS_PUBLIC_URL', { infer: true }), + }; + } + + private outboxIdempotencyKey( + run: DeployRunSummary, + event: DeployNotificationEvent, + ): string { + return [ + run.id, + `deploy.${event}`, + run.status, + run.jenkinsBuildNumber ?? run.jenkinsQueueId ?? 'no-jenkins-ref', + ].join(':'); + } + + private safeRefForOutbox(ref: string): string { + if (/^[a-f0-9]{40}$/i.test(ref)) { + return ref.slice(0, 12); + } + + return ref.length > 48 ? `${ref.slice(0, 45)}...` : ref; + } + private markdownFor( run: DeployRunSummary, event: DeployNotificationEvent, diff --git a/src/notifications/wecom/wecom.module.ts b/src/notifications/wecom/wecom.module.ts index e3a4f89..1b58161 100644 --- a/src/notifications/wecom/wecom.module.ts +++ b/src/notifications/wecom/wecom.module.ts @@ -1,8 +1,13 @@ +/** + * 通知模块集中注册 provider 客户端、发布通知服务和 outbox 能力。 + */ import { Module } from '@nestjs/common'; +import { NotificationOutboxModule } from '../outbox/notification-outbox.module'; import { DeployNotificationService } from './deploy-notification.service'; import { WeComClient } from './wecom.client'; @Module({ + imports: [NotificationOutboxModule], providers: [WeComClient, DeployNotificationService], exports: [WeComClient, DeployNotificationService], })