feat(M-AI-02-07): transactional OutboxEvent model
- 16 columns, dedupeKey UNIQUE, (status,availableAt) + (lockedAt) indexes - CAS markProcessing (updateMany where status='pending') - createInTransaction(tx) for atomic Job+Outbox writes - 12 unit tests (CAS, transaction, lock release, constraints) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wangdl
parent
7cc6947e41
commit
0dab626cd8
@ -0,0 +1,27 @@
|
||||
-- M-AI-02-07: 新增 Transactional Outbox 数据模型
|
||||
-- 用于可靠发布 BullMQ Job 和领域事件（本批不实现 Dispatcher）
|
||||
|
||||
CREATE TABLE `OutboxEvent` (
|
||||
`id` VARCHAR(191) NOT NULL,
|
||||
`eventType` VARCHAR(128) NOT NULL,
|
||||
`aggregateType` VARCHAR(64) NOT NULL,
|
||||
`aggregateId` VARCHAR(255) NOT NULL,
|
||||
`dedupeKey` VARCHAR(255) NOT NULL,
|
||||
`payload` JSON NOT NULL,
|
||||
`status` VARCHAR(32) NOT NULL DEFAULT 'pending',
|
||||
`attemptCount` INT NOT NULL DEFAULT 0,
|
||||
`availableAt` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
`lockedAt` DATETIME(3) NULL,
|
||||
`lockedBy` VARCHAR(100) NULL,
|
||||
`publishedAt` DATETIME(3) NULL,
|
||||
`lastErrorCode` VARCHAR(100) NULL,
|
||||
`lastErrorMessage` TEXT NULL,
|
||||
`createdAt` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
|
||||
`updatedAt` DATETIME(3) NOT NULL,
|
||||
|
||||
UNIQUE INDEX `OutboxEvent_dedupeKey_key` (`dedupeKey`),
|
||||
INDEX `OutboxEvent_status_availableAt_idx` (`status`, `availableAt`),
|
||||
INDEX `OutboxEvent_lockedAt_idx` (`lockedAt`),
|
||||
INDEX `OutboxEvent_aggregateType_aggregateId_idx` (`aggregateType`, `aggregateId`),
|
||||
PRIMARY KEY (`id`)
|
||||
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||||
178
src/infrastructure/outbox/outbox.repository.spec.ts
Normal file
178
src/infrastructure/outbox/outbox.repository.spec.ts
Normal file
@ -0,0 +1,178 @@
|
||||
import { Test, TestingModule } from '@nestjs/testing';
|
||||
import { Prisma } from '@prisma/client';
|
||||
import { OutboxRepository, CreateOutboxInput } from './outbox.repository';
|
||||
import { PrismaService } from '../database/prisma.service';
|
||||
|
||||
describe('OutboxRepository', () => {
|
||||
let repo: OutboxRepository;
|
||||
|
||||
const mockOutboxCreate = jest.fn();
|
||||
const mockOutboxFindMany = jest.fn();
|
||||
const mockOutboxFindUnique = jest.fn();
|
||||
const mockOutboxUpdate = jest.fn();
|
||||
const mockOutboxUpdateMany = jest.fn();
|
||||
|
||||
const mockTx = {
|
||||
outboxEvent: {
|
||||
create: mockOutboxCreate,
|
||||
},
|
||||
} as any;
|
||||
|
||||
beforeEach(async () => {
|
||||
mockOutboxCreate.mockReset();
|
||||
mockOutboxFindMany.mockReset();
|
||||
mockOutboxFindUnique.mockReset();
|
||||
mockOutboxUpdate.mockReset();
|
||||
mockOutboxUpdateMany.mockReset();
|
||||
|
||||
const module: TestingModule = await Test.createTestingModule({
|
||||
providers: [
|
||||
OutboxRepository,
|
||||
{
|
||||
provide: PrismaService,
|
||||
useValue: {
|
||||
outboxEvent: {
|
||||
create: mockOutboxCreate,
|
||||
findMany: mockOutboxFindMany,
|
||||
findUnique: mockOutboxFindUnique,
|
||||
update: mockOutboxUpdate,
|
||||
updateMany: mockOutboxUpdateMany,
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
}).compile();
|
||||
|
||||
repo = module.get(OutboxRepository);
|
||||
});
|
||||
|
||||
const sampleInput: CreateOutboxInput = {
|
||||
eventType: 'ai.job.completed',
|
||||
aggregateType: 'AiJob',
|
||||
aggregateId: 'job-1',
|
||||
dedupeKey: 'job-1:completed:v1',
|
||||
payload: { jobId: 'job-1', status: 'succeeded' },
|
||||
};
|
||||
|
||||
describe('createInTransaction', () => {
|
||||
it('应在事务中创建 Outbox 事件', async () => {
|
||||
mockOutboxCreate.mockResolvedValue({ id: 'evt-1', ...sampleInput, status: 'pending', createdAt: new Date() });
|
||||
|
||||
const result = await repo.createInTransaction(mockTx, sampleInput);
|
||||
expect(result).toHaveProperty('id', 'evt-1');
|
||||
expect(mockOutboxCreate).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('事务回滚时 Outbox 不存在（duplicate key 测试）', async () => {
|
||||
// 模拟：事务中先写 Job，再写 Outbox 时遇到 dedupeKey 冲突
|
||||
// 事务整体回滚，Job 也不存在
|
||||
const p2002 = new Prisma.PrismaClientKnownRequestError('dup', {
|
||||
code: 'P2002', clientVersion: '5.22.0', meta: {},
|
||||
});
|
||||
mockOutboxCreate.mockRejectedValueOnce(p2002);
|
||||
|
||||
await expect(repo.createInTransaction(mockTx, sampleInput)).rejects.toThrow();
|
||||
// 调用方应回滚整个事务
|
||||
});
|
||||
});
|
||||
|
||||
describe('findDispatchable', () => {
|
||||
it('应返回 pending 且 availableAt 已到的事件', async () => {
|
||||
mockOutboxFindMany.mockResolvedValue([]);
|
||||
await repo.findDispatchable(20);
|
||||
expect(mockOutboxFindMany).toHaveBeenCalledWith({
|
||||
where: { status: 'pending', availableAt: { lte: expect.any(Date) } },
|
||||
orderBy: { availableAt: 'asc' },
|
||||
take: 20,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('markProcessing（CAS）', () => {
|
||||
it('CAS 成功（status=pending）应获取锁并返回记录', async () => {
|
||||
mockOutboxUpdateMany.mockResolvedValue({ count: 1 });
|
||||
mockOutboxFindUnique.mockResolvedValue({ id: 'evt-1', status: 'processing', lockedBy: 'worker-1' });
|
||||
|
||||
const result = await repo.markProcessing('evt-1', 'worker-1');
|
||||
expect(result).toHaveProperty('status', 'processing');
|
||||
expect(mockOutboxUpdateMany).toHaveBeenCalledWith({
|
||||
where: { id: 'evt-1', status: 'pending' },
|
||||
data: { status: 'processing', lockedAt: expect.any(Date), lockedBy: 'worker-1' },
|
||||
});
|
||||
});
|
||||
|
||||
it('CAS 失败（已被其他 Worker 抢占）应返回 null', async () => {
|
||||
mockOutboxUpdateMany.mockResolvedValue({ count: 0 }); // 另一个 Worker 已改 status
|
||||
|
||||
const result = await repo.markProcessing('evt-1', 'worker-2');
|
||||
expect(result).toBeNull();
|
||||
expect(mockOutboxUpdateMany).toHaveBeenCalledTimes(1);
|
||||
expect(mockOutboxFindUnique).not.toHaveBeenCalled(); // 未获得锁，不查询
|
||||
});
|
||||
});
|
||||
|
||||
describe('markPublished', () => {
|
||||
it('应设置 status=published, publishedAt', async () => {
|
||||
mockOutboxUpdate.mockResolvedValue({ id: 'evt-1', status: 'published' });
|
||||
const result = await repo.markPublished('evt-1');
|
||||
expect(result).toHaveProperty('status', 'published');
|
||||
expect(mockOutboxUpdate).toHaveBeenCalledWith({
|
||||
where: { id: 'evt-1' },
|
||||
data: expect.objectContaining({ status: 'published' }),
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('markFailed', () => {
|
||||
it('应设置 status=failed, 递增 attemptCount', async () => {
|
||||
mockOutboxUpdate.mockResolvedValue({ id: 'evt-1', status: 'failed' });
|
||||
const result = await repo.markFailed('evt-1', 'PUBLISH_ERR', 'timeout');
|
||||
expect(result).toHaveProperty('status', 'failed');
|
||||
expect(mockOutboxUpdate).toHaveBeenCalledWith({
|
||||
where: { id: 'evt-1' },
|
||||
data: expect.objectContaining({
|
||||
status: 'failed',
|
||||
lastErrorCode: 'PUBLISH_ERR',
|
||||
lastErrorMessage: 'timeout',
|
||||
attemptCount: { increment: 1 },
|
||||
}),
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('releaseExpiredLocks', () => {
|
||||
it('应释放超过阈值的 processing 事件锁', async () => {
|
||||
mockOutboxUpdateMany.mockResolvedValue({ count: 3 });
|
||||
const count = await repo.releaseExpiredLocks(60_000);
|
||||
expect(count).toBe(3);
|
||||
expect(mockOutboxUpdateMany).toHaveBeenCalledWith({
|
||||
where: { status: 'processing', lockedAt: { lt: expect.any(Date) } },
|
||||
data: { status: 'pending', lockedAt: null, lockedBy: null },
|
||||
});
|
||||
});
|
||||
|
||||
it('应返回 0 当无过期锁', async () => {
|
||||
mockOutboxUpdateMany.mockResolvedValue({ count: 0 });
|
||||
const count = await repo.releaseExpiredLocks();
|
||||
expect(count).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('约束验证', () => {
|
||||
it('不应暴露 publish/publishToQueue 方法', () => {
|
||||
expect(typeof (repo as any).publish).toBe('undefined');
|
||||
expect(typeof (repo as any).publishToQueue).toBe('undefined');
|
||||
expect(typeof (repo as any).sendToBullMQ).toBe('undefined');
|
||||
});
|
||||
|
||||
it('不应暴露独立事务方法', () => {
|
||||
expect(typeof (repo as any).create).toBe('undefined');
|
||||
// createInTransaction 必须接受 tx 参数，不允许无 tx 调用
|
||||
});
|
||||
|
||||
it('dedupeKey UNIQUE 由 Prisma schema 保证', () => {
|
||||
// OutboxEvent.dedupeKey @unique — 数据库层保证
|
||||
expect(repo).toBeDefined();
|
||||
});
|
||||
});
|
||||
});
|
||||
139
src/infrastructure/outbox/outbox.repository.ts
Normal file
139
src/infrastructure/outbox/outbox.repository.ts
Normal file
@ -0,0 +1,139 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { PrismaService } from '../database/prisma.service';
|
||||
import { Prisma } from '@prisma/client';
|
||||
|
||||
export const OUTBOX_STATUSES = ['pending', 'processing', 'published', 'failed'] as const;
|
||||
export type OutboxStatus = (typeof OUTBOX_STATUSES)[number];
|
||||
|
||||
export interface CreateOutboxInput {
|
||||
eventType: string;
|
||||
aggregateType: string;
|
||||
aggregateId: string;
|
||||
dedupeKey: string;
|
||||
payload: Record<string, unknown>;
|
||||
availableAt?: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* M-AI-02-07: Transactional Outbox Repository
|
||||
*
|
||||
* 约束：
|
||||
* - createInTransaction 支持在外部 Prisma Transaction 中原子写入
|
||||
* - 不自行开启独立事务
|
||||
* - 不向 BullMQ 发布
|
||||
* - dedupeKey UNIQUE 保证幂等
|
||||
*/
|
||||
@Injectable()
|
||||
export class OutboxRepository {
|
||||
constructor(private readonly prisma: PrismaService) {}
|
||||
|
||||
/**
|
||||
* 在给定的事务客户端中创建 Outbox 事件。
|
||||
* 若未传 tx，使用 this.prisma（非事务模式，仅测试用）。
|
||||
*
|
||||
* 调用方负责在同一个 tx 中同时写入业务数据，
|
||||
* 以确保 Job + Outbox 的原子性。
|
||||
*/
|
||||
async createInTransaction(
|
||||
tx: Prisma.TransactionClient,
|
||||
input: CreateOutboxInput,
|
||||
) {
|
||||
return tx.outboxEvent.create({
|
||||
data: {
|
||||
eventType: input.eventType,
|
||||
aggregateType: input.aggregateType,
|
||||
aggregateId: input.aggregateId,
|
||||
dedupeKey: input.dedupeKey,
|
||||
payload: input.payload as any,
|
||||
availableAt: input.availableAt ?? new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 查找可分发的事件：status=pending 且 availableAt 已到。
|
||||
* 按 availableAt ASC 保证 FIFO。
|
||||
*/
|
||||
async findDispatchable(limit: number = 50) {
|
||||
return this.prisma.outboxEvent.findMany({
|
||||
where: {
|
||||
status: 'pending',
|
||||
availableAt: { lte: new Date() },
|
||||
},
|
||||
orderBy: { availableAt: 'asc' },
|
||||
take: limit,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 标记为 processing（CAS：仅当 status 仍为 'pending' 时成功）。
|
||||
* 返回更新后的记录；若被其他 Worker 抢占则返回 null。
|
||||
*/
|
||||
async markProcessing(eventId: string, lockedBy: string) {
|
||||
const result = await this.prisma.outboxEvent.updateMany({
|
||||
where: {
|
||||
id: eventId,
|
||||
status: 'pending',
|
||||
},
|
||||
data: {
|
||||
status: 'processing',
|
||||
lockedAt: new Date(),
|
||||
lockedBy,
|
||||
},
|
||||
});
|
||||
|
||||
if (result.count === 0) return null;
|
||||
|
||||
return this.prisma.outboxEvent.findUnique({
|
||||
where: { id: eventId },
|
||||
});
|
||||
}
|
||||
|
||||
/** 标记为 published */
|
||||
async markPublished(eventId: string) {
|
||||
return this.prisma.outboxEvent.update({
|
||||
where: { id: eventId },
|
||||
data: {
|
||||
status: 'published',
|
||||
publishedAt: new Date(),
|
||||
lockedAt: null,
|
||||
lockedBy: null,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/** 标记为 failed，递增 attemptCount */
|
||||
async markFailed(eventId: string, errorCode?: string, errorMessage?: string) {
|
||||
return this.prisma.outboxEvent.update({
|
||||
where: { id: eventId },
|
||||
data: {
|
||||
status: 'failed',
|
||||
lastErrorCode: errorCode,
|
||||
lastErrorMessage: errorMessage,
|
||||
attemptCount: { increment: 1 },
|
||||
lockedAt: null,
|
||||
lockedBy: null,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 释放过期锁：将 lockedAt 超过 thresholdMs 的 processing 事件重置为 pending。
|
||||
* 返回释放数量。
|
||||
*/
|
||||
async releaseExpiredLocks(thresholdMs: number = 60_000) {
|
||||
const cutoff = new Date(Date.now() - thresholdMs);
|
||||
const result = await this.prisma.outboxEvent.updateMany({
|
||||
where: {
|
||||
status: 'processing',
|
||||
lockedAt: { lt: cutoff },
|
||||
},
|
||||
data: {
|
||||
status: 'pending',
|
||||
lockedAt: null,
|
||||
lockedBy: null,
|
||||
},
|
||||
});
|
||||
return result.count;
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user