feat(wallet-service): 实现 Outbox Pattern 保证系统划转事件发布的可靠性
实现内容： - 添加 OutboxEvent 模型到 schema.prisma - 创建 OutboxRepository 服务处理事件持久化 - 创建 OutboxPublisherService 后台轮询发布事件到 Kafka - 修改 SystemWithdrawalApplicationService 将事件写入事务内 - 添加数据库迁移文件创建 outbox_events 表 技术细节： - 业务数据和事件数据在同一个数据库事务中写入 - 后台任务每秒轮询 outbox_events 表，发布 PENDING 状态事件 - 事件发布后标记为 SENT，等待消费方确认后标记为 CONFIRMED - 超时未确认的事件自动重试（指数退避策略） - 保证事件不丢失，即使 Kafka 暂时不可用 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
28f1e26400
commit
4f2f808484
|
|
@ -0,0 +1,53 @@
|
|||
-- ============================================
|
||||
-- Outbox 事件表 (可靠事件发布)
|
||||
-- [2026-01-07] 新增：实现 Outbox Pattern 确保事件发布的原子性
|
||||
--
|
||||
-- 用途：
|
||||
-- 1. 业务操作和事件写入在同一个事务中完成
|
||||
-- 2. 后台任务轮询此表发布事件到 Kafka
|
||||
-- 3. 保证事件不丢失，即使 Kafka 暂时不可用
|
||||
-- ============================================
|
||||
|
||||
CREATE TABLE IF NOT EXISTS "outbox_events" (
|
||||
"outbox_id" BIGSERIAL PRIMARY KEY,
|
||||
|
||||
-- 事件信息
|
||||
"event_type" VARCHAR(100) NOT NULL, -- 事件类型 (如 wallet.system-withdrawal.requested)
|
||||
"topic" VARCHAR(100) NOT NULL, -- Kafka topic
|
||||
"key" VARCHAR(200) NOT NULL, -- Kafka message key (通常是 orderNo 或 aggregateId)
|
||||
"payload" JSONB NOT NULL, -- 事件载荷 (JSON 格式)
|
||||
|
||||
-- 聚合根信息 (用于幂等性检查和事件确认)
|
||||
"aggregate_id" VARCHAR(100) NOT NULL, -- 聚合根ID (如 orderNo)
|
||||
"aggregate_type" VARCHAR(50) NOT NULL, -- 聚合根类型 (如 SystemWithdrawalOrder)
|
||||
|
||||
-- 发布状态
|
||||
"status" VARCHAR(20) NOT NULL DEFAULT 'PENDING', -- PENDING, SENT, CONFIRMED, FAILED
|
||||
"retry_count" INT NOT NULL DEFAULT 0, -- 重试次数
|
||||
"max_retries" INT NOT NULL DEFAULT 5, -- 最大重试次数
|
||||
"last_error" TEXT, -- 最后一次错误信息
|
||||
|
||||
-- 时间戳
|
||||
"created_at" TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, -- 创建时间
|
||||
"published_at" TIMESTAMP, -- 发布到 Kafka 的时间
|
||||
"next_retry_at" TIMESTAMP -- 下次重试时间 (失败时设置)
|
||||
);
|
||||
|
||||
-- 索引
|
||||
-- 1. 状态+创建时间：用于轮询待发布事件
|
||||
CREATE INDEX "idx_outbox_status_created" ON "outbox_events" ("status", "created_at");
|
||||
|
||||
-- 2. 状态+下次重试时间：用于查找需要重试的事件
|
||||
CREATE INDEX "idx_outbox_status_next_retry" ON "outbox_events" ("status", "next_retry_at");
|
||||
|
||||
-- 3. 聚合根类型+ID：用于按聚合根查找事件（确认时使用）
|
||||
CREATE INDEX "idx_outbox_aggregate" ON "outbox_events" ("aggregate_type", "aggregate_id");
|
||||
|
||||
-- 4. Topic：用于按主题统计
|
||||
CREATE INDEX "idx_outbox_topic" ON "outbox_events" ("topic");
|
||||
|
||||
-- 添加注释
|
||||
COMMENT ON TABLE "outbox_events" IS 'Outbox 事件表 - 实现可靠的事件发布机制';
|
||||
COMMENT ON COLUMN "outbox_events"."status" IS 'PENDING=待发送, SENT=已发送待确认, CONFIRMED=已确认, FAILED=失败';
|
||||
COMMENT ON COLUMN "outbox_events"."retry_count" IS '当前重试次数，达到 max_retries 后停止重试';
|
||||
COMMENT ON COLUMN "outbox_events"."next_retry_at" IS '使用指数退避策略计算的下次重试时间';
|
||||
|
|
@ -463,3 +463,38 @@ model OfflineSettlementDeduction {
|
|||
@@index([deductionOrderNo])
|
||||
@@index([createdAt])
|
||||
}
|
||||
|
||||
// ============================================
|
||||
// Outbox 事件表 (可靠事件发布)
|
||||
// [2026-01-07] 新增：实现 Outbox Pattern 确保事件发布的原子性
|
||||
// ============================================
|
||||
model OutboxEvent {
|
||||
id BigInt @id @default(autoincrement()) @map("outbox_id")
|
||||
|
||||
// 事件信息
|
||||
eventType String @map("event_type") @db.VarChar(100)
|
||||
topic String @map("topic") @db.VarChar(100)
|
||||
key String @map("key") @db.VarChar(200)
|
||||
payload Json @map("payload")
|
||||
|
||||
// 聚合根信息 (用于幂等性检查和事件确认)
|
||||
aggregateId String @map("aggregate_id") @db.VarChar(100)
|
||||
aggregateType String @map("aggregate_type") @db.VarChar(50)
|
||||
|
||||
// 发布状态: PENDING(待发送), SENT(已发送待确认), CONFIRMED(已确认), FAILED(失败)
|
||||
status String @default("PENDING") @map("status") @db.VarChar(20)
|
||||
retryCount Int @default(0) @map("retry_count")
|
||||
maxRetries Int @default(5) @map("max_retries")
|
||||
lastError String? @map("last_error") @db.Text
|
||||
|
||||
// 时间戳
|
||||
createdAt DateTime @default(now()) @map("created_at")
|
||||
publishedAt DateTime? @map("published_at")
|
||||
nextRetryAt DateTime? @map("next_retry_at")
|
||||
|
||||
@@index([status, createdAt])
|
||||
@@index([status, nextRetryAt])
|
||||
@@index([aggregateType, aggregateId])
|
||||
@@index([topic])
|
||||
@@map("outbox_events")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,14 +5,23 @@
|
|||
* - 支持从固定系统账户（总部、运营、积分股池等）转出
|
||||
* - 支持从区域账户（省区域、市区域）转出
|
||||
* - 记录双边流水（系统账户转出 + 用户账户转入）
|
||||
*
|
||||
* [2026-01-07] 更新：使用 Outbox Pattern 确保事件发布的原子性
|
||||
* - 事件写入 outbox_events 表与业务数据在同一个事务中
|
||||
* - 后台任务轮询 outbox 表发布事件到 Kafka
|
||||
* - 保证事件不丢失，即使 Kafka 暂时不可用
|
||||
*/
|
||||
|
||||
import { Injectable, Logger, BadRequestException, Inject } from '@nestjs/common';
|
||||
import { Injectable, Logger, BadRequestException } from '@nestjs/common';
|
||||
import { PrismaService } from '@/infrastructure/persistence/prisma/prisma.service';
|
||||
import { EventPublisherService } from '@/infrastructure/kafka';
|
||||
import { IdentityClientService } from '@/infrastructure/external/identity/identity-client.service';
|
||||
import { HotWalletCacheService } from '@/infrastructure/redis';
|
||||
import { LedgerEntryType } from '@/domain/value-objects/ledger-entry-type.enum';
|
||||
// [2026-01-07] 新增：使用 Outbox Pattern
|
||||
import {
|
||||
OutboxRepository,
|
||||
OutboxEventData,
|
||||
} from '@/infrastructure/persistence/repositories/outbox.repository';
|
||||
import Decimal from 'decimal.js';
|
||||
|
||||
// 系统账户名称映射
|
||||
|
|
@ -61,7 +70,8 @@ export class SystemWithdrawalApplicationService {
|
|||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly eventPublisher: EventPublisherService,
|
||||
// [2026-01-07] 更新：使用 OutboxRepository 替代直接的 EventPublisher
|
||||
private readonly outboxRepository: OutboxRepository,
|
||||
private readonly identityClient: IdentityClientService,
|
||||
private readonly hotWalletCacheService: HotWalletCacheService,
|
||||
) {}
|
||||
|
|
@ -229,6 +239,34 @@ export class SystemWithdrawalApplicationService {
|
|||
},
|
||||
});
|
||||
|
||||
// 6.6 [2026-01-07] 新增：在事务中写入 Outbox 事件
|
||||
// 使用 Outbox Pattern 确保事件发布的原子性
|
||||
const outboxEvent: OutboxEventData = {
|
||||
eventType: 'wallet.system-withdrawal.requested',
|
||||
topic: 'wallet.system-withdrawals',
|
||||
key: orderNo,
|
||||
aggregateId: orderNo,
|
||||
aggregateType: 'SystemWithdrawalOrder',
|
||||
payload: {
|
||||
eventId: `${Date.now()}-${Math.random().toString(36).substring(2, 9)}`,
|
||||
eventType: 'wallet.system-withdrawal.requested',
|
||||
occurredAt: new Date().toISOString(),
|
||||
payload: {
|
||||
orderNo,
|
||||
fromAccountSequence: command.fromAccountSequence,
|
||||
fromAccountName,
|
||||
toAccountSequence: command.toAccountSequence,
|
||||
toAddress: toUserInfo.walletAddress,
|
||||
amount: command.amount.toString(),
|
||||
chainType: 'KAVA',
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
await this.outboxRepository.saveInTransaction(tx, [outboxEvent]);
|
||||
|
||||
this.logger.log(`[SYSTEM_WITHDRAWAL] 转出订单和 Outbox 事件已写入事务: ${orderNo}`);
|
||||
|
||||
return {
|
||||
orderNo: order.orderNo,
|
||||
fromAccountSequence: command.fromAccountSequence,
|
||||
|
|
@ -241,21 +279,9 @@ export class SystemWithdrawalApplicationService {
|
|||
};
|
||||
});
|
||||
|
||||
// 7. 发布事件通知 blockchain-service 执行链上转账
|
||||
await this.eventPublisher.publish({
|
||||
eventType: 'wallet.system-withdrawal.requested',
|
||||
payload: {
|
||||
orderNo: result.orderNo,
|
||||
fromAccountSequence: result.fromAccountSequence,
|
||||
fromAccountName: result.fromAccountName,
|
||||
toAccountSequence: result.toAccountSequence,
|
||||
toAddress: result.toAddress,
|
||||
amount: command.amount.toString(),
|
||||
chainType: 'KAVA',
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.log(`[SYSTEM_WITHDRAWAL] 转出订单创建成功: ${result.orderNo}`);
|
||||
// 7. [2026-01-07] 更新：事件已在事务中写入 Outbox 表
|
||||
// 后台 OutboxPublisher 服务会自动轮询并发布到 Kafka
|
||||
this.logger.log(`[SYSTEM_WITHDRAWAL] 转出订单创建成功: ${result.orderNo}, 事件将由 OutboxPublisher 发布`);
|
||||
|
||||
return result;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,9 @@ import { EventPublisherService } from './event-publisher.service';
|
|||
import { DepositEventConsumerService } from './deposit-event-consumer.service';
|
||||
import { PlantingEventConsumerService } from './planting-event-consumer.service';
|
||||
import { WithdrawalEventConsumerService } from './withdrawal-event-consumer.service';
|
||||
// [2026-01-07] 新增：Outbox Pattern 实现
|
||||
import { OutboxPublisherService } from './outbox-publisher.service';
|
||||
import { OutboxRepository } from '../persistence/repositories/outbox.repository';
|
||||
// [已屏蔽] 前端直接从 reward-service 查询，不再订阅 reward-service 消息
|
||||
// import { RewardEventConsumerController } from './reward-event-consumer.controller';
|
||||
// import { EventAckPublisher } from './event-ack.publisher';
|
||||
|
|
@ -36,7 +39,25 @@ import { PrismaService } from '../persistence/prisma/prisma.service';
|
|||
// [已屏蔽] 前端直接从 reward-service 查询，不再订阅 reward-service 消息
|
||||
// controllers: [RewardEventConsumerController],
|
||||
controllers: [],
|
||||
providers: [PrismaService, EventPublisherService, DepositEventConsumerService, PlantingEventConsumerService, WithdrawalEventConsumerService],
|
||||
exports: [EventPublisherService, DepositEventConsumerService, PlantingEventConsumerService, WithdrawalEventConsumerService, ClientsModule],
|
||||
providers: [
|
||||
PrismaService,
|
||||
EventPublisherService,
|
||||
DepositEventConsumerService,
|
||||
PlantingEventConsumerService,
|
||||
WithdrawalEventConsumerService,
|
||||
// [2026-01-07] 新增：Outbox Pattern
|
||||
OutboxRepository,
|
||||
OutboxPublisherService,
|
||||
],
|
||||
exports: [
|
||||
EventPublisherService,
|
||||
DepositEventConsumerService,
|
||||
PlantingEventConsumerService,
|
||||
WithdrawalEventConsumerService,
|
||||
ClientsModule,
|
||||
// [2026-01-07] 新增：导出 Outbox 相关服务
|
||||
OutboxRepository,
|
||||
OutboxPublisherService,
|
||||
],
|
||||
})
|
||||
export class KafkaModule {}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,352 @@
|
|||
/**
|
||||
* Outbox Publisher Service
|
||||
* [2026-01-07] 新增：实现 Outbox Pattern 确保事件发布的原子性和可靠性
|
||||
*
|
||||
* 轮询 Outbox 表并发布事件到 Kafka
|
||||
* 使用消费方确认机制保证事件100%被处理
|
||||
*
|
||||
* 工作流程:
|
||||
* 1. 轮询 PENDING 状态的事件
|
||||
* 2. 发送到 Kafka，标记为 SENT（等待确认）
|
||||
* 3. 消费方处理成功后发送确认事件（如 blockchain.system-withdrawal.confirmed）
|
||||
* 4. 收到确认后标记为 CONFIRMED
|
||||
* 5. 超时未确认的事件重置为 PENDING 重发
|
||||
*/
|
||||
|
||||
import {
|
||||
Injectable,
|
||||
Logger,
|
||||
OnModuleInit,
|
||||
OnModuleDestroy,
|
||||
} from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { Kafka, Producer, logLevel } from 'kafkajs';
|
||||
import {
|
||||
OutboxRepository,
|
||||
OutboxEvent,
|
||||
} from '../persistence/repositories/outbox.repository';
|
||||
|
||||
@Injectable()
|
||||
export class OutboxPublisherService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(OutboxPublisherService.name);
|
||||
private kafka: Kafka;
|
||||
private producer: Producer;
|
||||
private isRunning = false;
|
||||
private pollInterval: NodeJS.Timeout | null = null;
|
||||
private timeoutCheckInterval: NodeJS.Timeout | null = null;
|
||||
private cleanupInterval: NodeJS.Timeout | null = null;
|
||||
private isConnected = false;
|
||||
|
||||
// 配置
|
||||
private readonly pollIntervalMs: number;
|
||||
private readonly batchSize: number;
|
||||
private readonly cleanupIntervalMs: number;
|
||||
private readonly confirmationTimeoutMinutes: number;
|
||||
private readonly timeoutCheckIntervalMs: number;
|
||||
|
||||
constructor(
|
||||
private readonly outboxRepository: OutboxRepository,
|
||||
private readonly configService: ConfigService,
|
||||
) {
|
||||
this.pollIntervalMs = this.configService.get<number>(
|
||||
'OUTBOX_POLL_INTERVAL_MS',
|
||||
1000,
|
||||
);
|
||||
this.batchSize = this.configService.get<number>('OUTBOX_BATCH_SIZE', 100);
|
||||
this.cleanupIntervalMs = this.configService.get<number>(
|
||||
'OUTBOX_CLEANUP_INTERVAL_MS',
|
||||
3600000,
|
||||
); // 1小时
|
||||
this.confirmationTimeoutMinutes = this.configService.get<number>(
|
||||
'OUTBOX_CONFIRMATION_TIMEOUT_MINUTES',
|
||||
5,
|
||||
);
|
||||
this.timeoutCheckIntervalMs = this.configService.get<number>(
|
||||
'OUTBOX_TIMEOUT_CHECK_INTERVAL_MS',
|
||||
60000,
|
||||
); // 1分钟
|
||||
|
||||
this.logger.log(
|
||||
`[OUTBOX] OutboxPublisher configured: ` +
|
||||
`pollInterval=${this.pollIntervalMs}ms, batchSize=${this.batchSize}, ` +
|
||||
`confirmationTimeout=${this.confirmationTimeoutMinutes}min`,
|
||||
);
|
||||
}
|
||||
|
||||
async onModuleInit() {
|
||||
const brokers =
|
||||
this.configService.get<string>('KAFKA_BROKERS')?.split(',') || [
|
||||
'localhost:9092',
|
||||
];
|
||||
const clientId =
|
||||
this.configService.get<string>('KAFKA_CLIENT_ID') || 'wallet-service';
|
||||
|
||||
this.logger.log('[OUTBOX] Connecting to Kafka...');
|
||||
|
||||
this.kafka = new Kafka({
|
||||
clientId: `${clientId}-outbox`,
|
||||
brokers,
|
||||
logLevel: logLevel.WARN,
|
||||
retry: {
|
||||
initialRetryTime: 100,
|
||||
retries: 8,
|
||||
},
|
||||
});
|
||||
|
||||
this.producer = this.kafka.producer();
|
||||
|
||||
try {
|
||||
await this.producer.connect();
|
||||
this.isConnected = true;
|
||||
this.logger.log('[OUTBOX] Connected to Kafka');
|
||||
this.start();
|
||||
} catch (error) {
|
||||
this.logger.error('[OUTBOX] Failed to connect to Kafka:', error);
|
||||
this.logger.warn(
|
||||
'[OUTBOX] OutboxPublisher will not start - events will accumulate in outbox table',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
this.stop();
|
||||
if (this.isConnected) {
|
||||
await this.producer.disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 启动轮询
|
||||
*/
|
||||
start(): void {
|
||||
if (this.isRunning) {
|
||||
this.logger.warn('[OUTBOX] Publisher already running');
|
||||
return;
|
||||
}
|
||||
|
||||
this.isRunning = true;
|
||||
this.logger.log('[OUTBOX] Starting outbox publisher...');
|
||||
|
||||
// 启动发布轮询
|
||||
this.pollInterval = setInterval(() => {
|
||||
this.processOutbox().catch((err) => {
|
||||
this.logger.error('[OUTBOX] Error processing outbox:', err);
|
||||
});
|
||||
}, this.pollIntervalMs);
|
||||
|
||||
// 启动超时检查任务
|
||||
this.timeoutCheckInterval = setInterval(() => {
|
||||
this.checkConfirmationTimeouts().catch((err) => {
|
||||
this.logger.error('[OUTBOX] Error checking confirmation timeouts:', err);
|
||||
});
|
||||
}, this.timeoutCheckIntervalMs);
|
||||
|
||||
// 启动清理任务
|
||||
this.cleanupInterval = setInterval(() => {
|
||||
this.cleanup().catch((err) => {
|
||||
this.logger.error('[OUTBOX] Error cleaning up outbox:', err);
|
||||
});
|
||||
}, this.cleanupIntervalMs);
|
||||
|
||||
this.logger.log('[OUTBOX] Outbox publisher started (消费方确认模式)');
|
||||
}
|
||||
|
||||
/**
|
||||
* 停止轮询
|
||||
*/
|
||||
stop(): void {
|
||||
if (!this.isRunning) return;
|
||||
|
||||
this.isRunning = false;
|
||||
|
||||
if (this.pollInterval) {
|
||||
clearInterval(this.pollInterval);
|
||||
this.pollInterval = null;
|
||||
}
|
||||
|
||||
if (this.timeoutCheckInterval) {
|
||||
clearInterval(this.timeoutCheckInterval);
|
||||
this.timeoutCheckInterval = null;
|
||||
}
|
||||
|
||||
if (this.cleanupInterval) {
|
||||
clearInterval(this.cleanupInterval);
|
||||
this.cleanupInterval = null;
|
||||
}
|
||||
|
||||
this.logger.log('[OUTBOX] Outbox publisher stopped');
|
||||
}
|
||||
|
||||
/**
|
||||
* 处理 Outbox 事件
|
||||
*/
|
||||
async processOutbox(): Promise<void> {
|
||||
if (!this.isConnected) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// 1. 获取待发布事件
|
||||
const pendingEvents =
|
||||
await this.outboxRepository.findPendingEvents(this.batchSize);
|
||||
|
||||
// 2. 获取需要重试的事件
|
||||
const retryEvents = await this.outboxRepository.findEventsForRetry(
|
||||
Math.floor(this.batchSize / 2),
|
||||
);
|
||||
|
||||
const allEvents = [...pendingEvents, ...retryEvents];
|
||||
|
||||
if (allEvents.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug(
|
||||
`[OUTBOX] Processing ${allEvents.length} events (${pendingEvents.length} pending, ${retryEvents.length} retry)`,
|
||||
);
|
||||
|
||||
// 3. 逐个发布
|
||||
for (const event of allEvents) {
|
||||
await this.publishEvent(event);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error('[OUTBOX] Error in processOutbox:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 发布单个事件
|
||||
*
|
||||
* 使用 producer.send() 发送到 Kafka，成功后标记为 SENT（等待消费方确认）
|
||||
* 只有收到消费方确认后才标记为 CONFIRMED
|
||||
*/
|
||||
private async publishEvent(event: OutboxEvent): Promise<void> {
|
||||
try {
|
||||
this.logger.debug(
|
||||
`[OUTBOX] Publishing event ${event.id} to topic ${event.topic}`,
|
||||
);
|
||||
|
||||
// 构造 Kafka 消息，包含 outbox 元数据用于确认
|
||||
const payload = {
|
||||
...(event.payload as Record<string, unknown>),
|
||||
_outbox: {
|
||||
id: event.id.toString(),
|
||||
aggregateId: event.aggregateId,
|
||||
eventType: event.eventType,
|
||||
},
|
||||
};
|
||||
|
||||
const message = {
|
||||
key: event.key,
|
||||
value: JSON.stringify(payload),
|
||||
headers: {
|
||||
eventType: event.eventType,
|
||||
source: 'wallet-service',
|
||||
outboxId: event.id.toString(),
|
||||
},
|
||||
};
|
||||
|
||||
// 发布到 Kafka
|
||||
await this.producer.send({
|
||||
topic: event.topic,
|
||||
messages: [message],
|
||||
});
|
||||
|
||||
// 标记为 SENT（等待消费方确认）
|
||||
await this.outboxRepository.markAsSent(event.id);
|
||||
|
||||
this.logger.log(
|
||||
`[OUTBOX] → Event ${event.id} sent to ${event.topic} (awaiting consumer confirmation)`,
|
||||
);
|
||||
} catch (error) {
|
||||
const errorMessage =
|
||||
error instanceof Error ? error.message : String(error);
|
||||
this.logger.error(
|
||||
`[OUTBOX] ✗ Failed to publish event ${event.id}: ${errorMessage}`,
|
||||
);
|
||||
|
||||
// 标记为失败并安排重试
|
||||
await this.outboxRepository.markAsFailed(event.id, errorMessage);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查确认超时的事件
|
||||
*
|
||||
* 将超时未确认的 SENT 事件重置为 PENDING 以便重发
|
||||
*/
|
||||
private async checkConfirmationTimeouts(): Promise<void> {
|
||||
if (!this.isConnected) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const timedOutEvents = await this.outboxRepository.findSentEventsTimedOut(
|
||||
this.confirmationTimeoutMinutes,
|
||||
this.batchSize,
|
||||
);
|
||||
|
||||
if (timedOutEvents.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.warn(
|
||||
`[OUTBOX] Found ${timedOutEvents.length} events without confirmation after ${this.confirmationTimeoutMinutes} minutes`,
|
||||
);
|
||||
|
||||
for (const event of timedOutEvents) {
|
||||
await this.outboxRepository.resetSentToPending(event.id);
|
||||
this.logger.warn(
|
||||
`[OUTBOX] Event ${event.id} reset to PENDING for retry (retry #${event.retryCount + 1})`,
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error('[OUTBOX] Error checking confirmation timeouts:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理旧事件
|
||||
*/
|
||||
private async cleanup(): Promise<void> {
|
||||
const retentionDays = this.configService.get<number>(
|
||||
'OUTBOX_RETENTION_DAYS',
|
||||
7,
|
||||
);
|
||||
await this.outboxRepository.cleanupOldEvents(retentionDays);
|
||||
}
|
||||
|
||||
/**
|
||||
* 手动触发处理（用于测试或紧急情况）
|
||||
*/
|
||||
async triggerProcessing(): Promise<void> {
|
||||
this.logger.log('[OUTBOX] Manual processing triggered');
|
||||
await this.processOutbox();
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取统计信息
|
||||
*/
|
||||
async getStats(): Promise<{
|
||||
isRunning: boolean;
|
||||
isConnected: boolean;
|
||||
pending: number;
|
||||
sent: number;
|
||||
confirmed: number;
|
||||
failed: number;
|
||||
}> {
|
||||
const stats = await this.outboxRepository.getStats();
|
||||
return {
|
||||
isRunning: this.isRunning,
|
||||
isConnected: this.isConnected,
|
||||
...stats,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 确认事件（供消费者确认处理器调用）
|
||||
*/
|
||||
async confirmEvent(aggregateId: string, eventType?: string): Promise<boolean> {
|
||||
return this.outboxRepository.markAsConfirmed(aggregateId, eventType);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,373 @@
|
|||
/**
|
||||
* Outbox Repository
|
||||
* [2026-01-07] 新增：实现 Outbox Pattern 确保事件发布的原子性和可靠性
|
||||
*
|
||||
* 事件流程：
|
||||
* 1. 业务操作和 Outbox 事件写入在同一个事务中完成
|
||||
* 2. 后台任务轮询 Outbox 表，发布事件到 Kafka
|
||||
* 3. 发布成功后标记为 SENT，等待消费方确认
|
||||
* 4. 收到消费方确认后标记为 CONFIRMED
|
||||
* 5. 失败的事件会按指数退避策略重试
|
||||
*/
|
||||
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { Prisma, PrismaClient } from '@prisma/client';
|
||||
|
||||
export enum OutboxStatus {
|
||||
PENDING = 'PENDING', // 待发送
|
||||
SENT = 'SENT', // 已发送到 Kafka，等待消费方确认
|
||||
CONFIRMED = 'CONFIRMED', // 消费方已确认处理成功
|
||||
FAILED = 'FAILED', // 发送失败，等待重试
|
||||
}
|
||||
|
||||
export interface OutboxEventData {
|
||||
eventType: string;
|
||||
topic: string;
|
||||
key: string;
|
||||
payload: Record<string, unknown>;
|
||||
aggregateId: string;
|
||||
aggregateType: string;
|
||||
}
|
||||
|
||||
export interface OutboxEvent extends OutboxEventData {
|
||||
id: bigint;
|
||||
status: OutboxStatus;
|
||||
retryCount: number;
|
||||
maxRetries: number;
|
||||
lastError: string | null;
|
||||
createdAt: Date;
|
||||
publishedAt: Date | null;
|
||||
nextRetryAt: Date | null;
|
||||
}
|
||||
|
||||
// Prisma 事务客户端类型（与 Prisma $transaction 返回的类型一致）
|
||||
export type PrismaTransactionClient = Omit<
|
||||
PrismaClient,
|
||||
'$connect' | '$disconnect' | '$on' | '$transaction' | '$use' | '$extends'
|
||||
>;
|
||||
|
||||
@Injectable()
|
||||
export class OutboxRepository {
|
||||
private readonly logger = new Logger(OutboxRepository.name);
|
||||
|
||||
constructor(private readonly prisma: PrismaService) {}
|
||||
|
||||
/**
|
||||
* 在事务中保存 Outbox 事件
|
||||
* 这是 Outbox Pattern 的核心：业务数据和事件数据在同一个事务中写入
|
||||
*/
|
||||
async saveInTransaction(
|
||||
tx: PrismaTransactionClient,
|
||||
events: OutboxEventData[],
|
||||
): Promise<void> {
|
||||
if (events.length === 0) return;
|
||||
|
||||
this.logger.debug(
|
||||
`[OUTBOX] Saving ${events.length} events to outbox (in transaction)`,
|
||||
);
|
||||
|
||||
await tx.outboxEvent.createMany({
|
||||
data: events.map((event) => ({
|
||||
eventType: event.eventType,
|
||||
topic: event.topic,
|
||||
key: event.key,
|
||||
payload: event.payload as Prisma.JsonObject,
|
||||
aggregateId: event.aggregateId,
|
||||
aggregateType: event.aggregateType,
|
||||
status: OutboxStatus.PENDING,
|
||||
})),
|
||||
});
|
||||
|
||||
this.logger.debug(`[OUTBOX] Saved ${events.length} events to outbox`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 直接保存 Outbox 事件（不在事务中）
|
||||
*/
|
||||
async saveEvents(events: OutboxEventData[]): Promise<void> {
|
||||
if (events.length === 0) return;
|
||||
|
||||
this.logger.debug(`[OUTBOX] Saving ${events.length} events to outbox`);
|
||||
|
||||
await this.prisma.outboxEvent.createMany({
|
||||
data: events.map((event) => ({
|
||||
eventType: event.eventType,
|
||||
topic: event.topic,
|
||||
key: event.key,
|
||||
payload: event.payload as Prisma.JsonObject,
|
||||
aggregateId: event.aggregateId,
|
||||
aggregateType: event.aggregateType,
|
||||
status: OutboxStatus.PENDING,
|
||||
})),
|
||||
});
|
||||
|
||||
this.logger.log(`[OUTBOX] ✓ Saved ${events.length} events to outbox`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取待发布的事件（按创建时间排序）
|
||||
*/
|
||||
async findPendingEvents(limit: number = 100): Promise<OutboxEvent[]> {
|
||||
const events = await this.prisma.outboxEvent.findMany({
|
||||
where: {
|
||||
status: OutboxStatus.PENDING,
|
||||
},
|
||||
orderBy: {
|
||||
createdAt: 'asc',
|
||||
},
|
||||
take: limit,
|
||||
});
|
||||
|
||||
return events.map((e) => this.mapToOutboxEvent(e));
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取需要重试的事件
|
||||
*/
|
||||
async findEventsForRetry(limit: number = 50): Promise<OutboxEvent[]> {
|
||||
const now = new Date();
|
||||
|
||||
const events = await this.prisma.outboxEvent.findMany({
|
||||
where: {
|
||||
status: OutboxStatus.FAILED,
|
||||
retryCount: {
|
||||
lt: 5, // maxRetries
|
||||
},
|
||||
nextRetryAt: {
|
||||
lte: now,
|
||||
},
|
||||
},
|
||||
orderBy: {
|
||||
nextRetryAt: 'asc',
|
||||
},
|
||||
take: limit,
|
||||
});
|
||||
|
||||
return events.map((e) => this.mapToOutboxEvent(e));
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记事件为已发送（等待消费方确认）
|
||||
*/
|
||||
async markAsSent(id: bigint): Promise<void> {
|
||||
await this.prisma.outboxEvent.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: OutboxStatus.SENT,
|
||||
publishedAt: new Date(),
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.debug(
|
||||
`[OUTBOX] Marked event ${id} as SENT (awaiting consumer confirmation)`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记事件为已确认（消费方已成功处理）
|
||||
* 使用 aggregateId + eventType 组合精确匹配
|
||||
*/
|
||||
async markAsConfirmed(
|
||||
aggregateId: string,
|
||||
eventType?: string,
|
||||
): Promise<boolean> {
|
||||
const whereClause: Prisma.OutboxEventWhereInput = {
|
||||
aggregateId: aggregateId,
|
||||
status: OutboxStatus.SENT,
|
||||
};
|
||||
|
||||
// 如果提供了 eventType，则精确匹配
|
||||
if (eventType) {
|
||||
whereClause.eventType = eventType;
|
||||
}
|
||||
|
||||
const result = await this.prisma.outboxEvent.updateMany({
|
||||
where: whereClause,
|
||||
data: {
|
||||
status: OutboxStatus.CONFIRMED,
|
||||
},
|
||||
});
|
||||
|
||||
if (result.count > 0) {
|
||||
this.logger.log(
|
||||
`[OUTBOX] ✓ Event ${aggregateId} (${eventType || 'all types'}) confirmed by consumer`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
this.logger.warn(
|
||||
`[OUTBOX] Event ${aggregateId} (${eventType || 'any'}) not found or not in SENT status`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过 outbox ID 标记为已确认
|
||||
*/
|
||||
async markAsConfirmedById(id: bigint): Promise<void> {
|
||||
await this.prisma.outboxEvent.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: OutboxStatus.CONFIRMED,
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.log(`[OUTBOX] ✓ Event ${id} confirmed by consumer`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取已发送但未确认且超时的事件（用于重试）
|
||||
*/
|
||||
async findSentEventsTimedOut(
|
||||
timeoutMinutes: number = 5,
|
||||
limit: number = 50,
|
||||
): Promise<OutboxEvent[]> {
|
||||
const cutoffTime = new Date();
|
||||
cutoffTime.setMinutes(cutoffTime.getMinutes() - timeoutMinutes);
|
||||
|
||||
const events = await this.prisma.outboxEvent.findMany({
|
||||
where: {
|
||||
status: OutboxStatus.SENT,
|
||||
publishedAt: {
|
||||
lt: cutoffTime,
|
||||
},
|
||||
retryCount: {
|
||||
lt: 5, // 最多重试5次
|
||||
},
|
||||
},
|
||||
orderBy: {
|
||||
publishedAt: 'asc',
|
||||
},
|
||||
take: limit,
|
||||
});
|
||||
|
||||
return events.map((e) => this.mapToOutboxEvent(e));
|
||||
}
|
||||
|
||||
/**
|
||||
* 将超时的 SENT 事件重置为 PENDING 以便重发
|
||||
*/
|
||||
async resetSentToPending(id: bigint): Promise<void> {
|
||||
const event = await this.prisma.outboxEvent.findUnique({
|
||||
where: { id },
|
||||
});
|
||||
|
||||
if (!event) return;
|
||||
|
||||
await this.prisma.outboxEvent.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: OutboxStatus.PENDING,
|
||||
retryCount: event.retryCount + 1,
|
||||
lastError: 'Consumer confirmation timeout',
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.warn(
|
||||
`[OUTBOX] Event ${id} reset to PENDING for retry (confirmation timeout)`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记事件为失败并安排重试
|
||||
*/
|
||||
async markAsFailed(id: bigint, error: string): Promise<void> {
|
||||
const event = await this.prisma.outboxEvent.findUnique({
|
||||
where: { id },
|
||||
});
|
||||
|
||||
if (!event) return;
|
||||
|
||||
const newRetryCount = event.retryCount + 1;
|
||||
const isFinalFailure = newRetryCount >= event.maxRetries;
|
||||
|
||||
// 指数退避：1min, 2min, 4min, 8min, 16min
|
||||
const delayMinutes = Math.pow(2, newRetryCount - 1);
|
||||
const nextRetryAt = new Date();
|
||||
nextRetryAt.setMinutes(nextRetryAt.getMinutes() + delayMinutes);
|
||||
|
||||
await this.prisma.outboxEvent.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: OutboxStatus.FAILED,
|
||||
retryCount: newRetryCount,
|
||||
lastError: error,
|
||||
nextRetryAt: isFinalFailure ? null : nextRetryAt,
|
||||
},
|
||||
});
|
||||
|
||||
if (isFinalFailure) {
|
||||
this.logger.error(
|
||||
`[OUTBOX] Event ${id} permanently failed after ${newRetryCount} retries: ${error}`,
|
||||
);
|
||||
} else {
|
||||
this.logger.warn(
|
||||
`[OUTBOX] Event ${id} failed, retry ${newRetryCount}/${event.maxRetries}, next retry at ${nextRetryAt}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 清理已确认的旧事件（保留最近7天）
|
||||
*/
|
||||
async cleanupOldEvents(retentionDays: number = 7): Promise<number> {
|
||||
const cutoffDate = new Date();
|
||||
cutoffDate.setDate(cutoffDate.getDate() - retentionDays);
|
||||
|
||||
const result = await this.prisma.outboxEvent.deleteMany({
|
||||
where: {
|
||||
status: OutboxStatus.CONFIRMED,
|
||||
publishedAt: {
|
||||
lt: cutoffDate,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
if (result.count > 0) {
|
||||
this.logger.log(`[OUTBOX] Cleaned up ${result.count} old events`);
|
||||
}
|
||||
|
||||
return result.count;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取 Outbox 统计信息
|
||||
*/
|
||||
async getStats(): Promise<{
|
||||
pending: number;
|
||||
sent: number;
|
||||
confirmed: number;
|
||||
failed: number;
|
||||
}> {
|
||||
const [pending, sent, confirmed, failed] = await Promise.all([
|
||||
this.prisma.outboxEvent.count({ where: { status: OutboxStatus.PENDING } }),
|
||||
this.prisma.outboxEvent.count({ where: { status: OutboxStatus.SENT } }),
|
||||
this.prisma.outboxEvent.count({
|
||||
where: { status: OutboxStatus.CONFIRMED },
|
||||
}),
|
||||
this.prisma.outboxEvent.count({ where: { status: OutboxStatus.FAILED } }),
|
||||
]);
|
||||
|
||||
return { pending, sent, confirmed, failed };
|
||||
}
|
||||
|
||||
private mapToOutboxEvent(record: any): OutboxEvent {
|
||||
return {
|
||||
id: record.id,
|
||||
eventType: record.eventType,
|
||||
topic: record.topic,
|
||||
key: record.key,
|
||||
payload: record.payload as Record<string, unknown>,
|
||||
aggregateId: record.aggregateId,
|
||||
aggregateType: record.aggregateType,
|
||||
status: record.status as OutboxStatus,
|
||||
retryCount: record.retryCount,
|
||||
maxRetries: record.maxRetries,
|
||||
lastError: record.lastError,
|
||||
createdAt: record.createdAt,
|
||||
publishedAt: record.publishedAt,
|
||||
nextRetryAt: record.nextRetryAt,
|
||||
};
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue