feat: 增加通知 outbox 重试器

- notifications: 支持失败 outbox 按配置 nextAttemptAt 重试并达到上限转 dead

- settings: 暴露通知 outbox 重试相关环境变量说明

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,

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<OutboxStatus, NotificationOutboxStatus> = {
@@ -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<NotificationOutboxMessage[]> {
+    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,

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;
+};

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<DeployNotificationResult> {
+    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<void> {
+    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<void> {
+    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<string, unknown> | 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;
+  }
 }

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<typeof setInterval> | undefined;
+  private running = false;
+
+  constructor(
+    private readonly config: ConfigService<EnvConfig, true>,
+    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<void> {
+    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;
+    }
+  }
+}

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 {}

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' },
+      ),
     },
   },
   {