459 lines
14 KiB
TypeScript
459 lines
14 KiB
TypeScript
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||
import type { OutboxEvent } from "../../src/shared/outbox/outbox.schema.js";
|
||
|
||
// ============================================================
|
||
// Mock 外部依赖 —— 使用 vi.hoisted 确保 mock 变量在 hoisted 的 vi.mock 中可用
|
||
// ============================================================
|
||
|
||
const mocks = vi.hoisted(() => {
|
||
const mockProducer = {
|
||
send: vi.fn(),
|
||
};
|
||
const mockFindPending = vi.fn();
|
||
const mockMarkPublished = vi.fn();
|
||
const mockMarkFailed = vi.fn();
|
||
const mockIncrementRetry = vi.fn();
|
||
return {
|
||
mockProducer,
|
||
mockFindPending,
|
||
mockMarkPublished,
|
||
mockMarkFailed,
|
||
mockIncrementRetry,
|
||
};
|
||
});
|
||
|
||
// Mock logger
|
||
vi.mock("../../src/shared/observability/logger.js", () => ({
|
||
logger: {
|
||
info: vi.fn(),
|
||
warn: vi.fn(),
|
||
error: vi.fn(),
|
||
debug: vi.fn(),
|
||
},
|
||
}));
|
||
|
||
// Mock Kafka client
|
||
vi.mock("../../src/shared/kafka/kafka.client.js", () => ({
|
||
getProducer: () => mocks.mockProducer,
|
||
isKafkaHealthy: vi.fn(() => true),
|
||
}));
|
||
|
||
// Mock topic-map
|
||
vi.mock("../../src/shared/kafka/topic-map.js", () => ({
|
||
resolveTopic: vi.fn((eventType: string) => {
|
||
const map: Record<string, string> = {
|
||
"notification.sent": "edu.notification.sent",
|
||
"notification.read": "edu.notification.read",
|
||
"notification.recalled": "edu.notification.recalled",
|
||
"notification.failed": "edu.notification.failed",
|
||
};
|
||
return map[eventType] ?? "edu.notification.events";
|
||
}),
|
||
}));
|
||
|
||
// Mock outbox repository
|
||
vi.mock("../../src/shared/outbox/outbox.repository.js", () => ({
|
||
findPending: mocks.mockFindPending,
|
||
markPublished: mocks.mockMarkPublished,
|
||
markFailed: mocks.mockMarkFailed,
|
||
incrementRetry: mocks.mockIncrementRetry,
|
||
}));
|
||
|
||
// 导入被测模块（在 mock 之后）
|
||
import { outboxPublisher } from "../../src/shared/outbox/outbox.publisher.js";
|
||
|
||
// ============================================================
|
||
// 辅助
|
||
// ============================================================
|
||
|
||
const {
|
||
mockProducer,
|
||
mockFindPending,
|
||
mockMarkPublished,
|
||
mockMarkFailed,
|
||
mockIncrementRetry,
|
||
} = mocks;
|
||
|
||
function createMessage(overrides: Partial<OutboxEvent> = {}): OutboxEvent {
|
||
return {
|
||
eventId: "evt-1",
|
||
aggregateType: "Notification",
|
||
aggregateId: "agg-1",
|
||
eventType: "notification.sent",
|
||
topic: "edu.notification.sent",
|
||
payload: { notificationId: "n1", userId: "u1" },
|
||
status: "pending",
|
||
retryCount: 0,
|
||
maxRetryCount: 5,
|
||
createdAt: new Date("2026-01-01T00:00:00Z"),
|
||
publishedAt: null,
|
||
nextRetryAt: null,
|
||
lastError: null,
|
||
metadata: null,
|
||
...overrides,
|
||
} as OutboxEvent;
|
||
}
|
||
|
||
// 访问私有方法
|
||
function poll(): Promise<void> {
|
||
return (outboxPublisher as unknown as { poll: () => Promise<void> }).poll();
|
||
}
|
||
|
||
function dispatch(message: OutboxEvent): Promise<void> {
|
||
return (
|
||
outboxPublisher as unknown as {
|
||
dispatch: (m: OutboxEvent) => Promise<void>;
|
||
}
|
||
).dispatch(message);
|
||
}
|
||
|
||
// ============================================================
|
||
// Tests
|
||
// ============================================================
|
||
|
||
describe("OutboxPublisher", () => {
|
||
beforeEach(async () => {
|
||
vi.clearAllMocks();
|
||
vi.useFakeTimers();
|
||
// 重置 singleton 状态：确保 intervalId 被清除
|
||
await outboxPublisher.stop();
|
||
mockProducer.send.mockResolvedValue({} as never);
|
||
mockMarkPublished.mockResolvedValue(undefined);
|
||
mockMarkFailed.mockResolvedValue(undefined);
|
||
mockIncrementRetry.mockResolvedValue(undefined);
|
||
mockFindPending.mockResolvedValue([]);
|
||
});
|
||
|
||
afterEach(() => {
|
||
vi.useRealTimers();
|
||
});
|
||
|
||
// ----------------------------------------------------------
|
||
// start / stop
|
||
// ----------------------------------------------------------
|
||
describe("start / stop", () => {
|
||
it("start 应设置定时轮询", async () => {
|
||
await outboxPublisher.start();
|
||
|
||
// 推进定时器触发 poll
|
||
await vi.advanceTimersByTimeAsync(5000);
|
||
|
||
expect(mockFindPending).toHaveBeenCalled();
|
||
});
|
||
|
||
it("重复 start 不应创建多个定时器", async () => {
|
||
await outboxPublisher.start();
|
||
await outboxPublisher.start();
|
||
|
||
// 推进 5 秒，应只 poll 一次（第二个 start 是 no-op）
|
||
await vi.advanceTimersByTimeAsync(5000);
|
||
|
||
// 只有一个 interval，poll 应只被调用一次
|
||
expect(mockFindPending).toHaveBeenCalledTimes(1);
|
||
});
|
||
|
||
it("stop 应清除定时器", async () => {
|
||
await outboxPublisher.start();
|
||
await outboxPublisher.stop();
|
||
|
||
mockFindPending.mockClear();
|
||
await vi.advanceTimersByTimeAsync(10000);
|
||
|
||
expect(mockFindPending).not.toHaveBeenCalled();
|
||
});
|
||
|
||
it("无定时器时 stop 不应报错", async () => {
|
||
await outboxPublisher.stop();
|
||
// 不抛出即可
|
||
});
|
||
});
|
||
|
||
// ----------------------------------------------------------
|
||
// poll
|
||
// ----------------------------------------------------------
|
||
describe("poll", () => {
|
||
it("findPending 返回消息时应逐条 dispatch", async () => {
|
||
const messages = [
|
||
createMessage({ eventId: "evt-1" }),
|
||
createMessage({ eventId: "evt-2" }),
|
||
];
|
||
mockFindPending.mockResolvedValue(messages);
|
||
|
||
await poll();
|
||
|
||
expect(mockFindPending).toHaveBeenCalledWith(100); // BATCH_SIZE
|
||
expect(mockProducer.send).toHaveBeenCalledTimes(2);
|
||
expect(mockMarkPublished).toHaveBeenCalledTimes(2);
|
||
expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
|
||
expect(mockMarkPublished).toHaveBeenCalledWith("evt-2");
|
||
});
|
||
|
||
it("findPending 返回空数组时不应 dispatch", async () => {
|
||
mockFindPending.mockResolvedValue([]);
|
||
|
||
await poll();
|
||
|
||
expect(mockProducer.send).not.toHaveBeenCalled();
|
||
expect(mockMarkPublished).not.toHaveBeenCalled();
|
||
});
|
||
|
||
it("findPending 抛出异常时应记录日志不中断", async () => {
|
||
mockFindPending.mockRejectedValue(new Error("DB connection lost"));
|
||
|
||
// 不应抛出
|
||
await poll();
|
||
|
||
// logger.error 应被调用（由 mock 拦截）
|
||
});
|
||
|
||
it("并发 poll 保护：正在 poll 时不应重复执行", async () => {
|
||
const messages = [createMessage()];
|
||
// 让 producer.send 返回一个未完成的 Promise
|
||
let resolveSend: () => void;
|
||
mockProducer.send.mockReturnValue(
|
||
new Promise((resolve) => {
|
||
resolveSend = resolve as () => void;
|
||
}),
|
||
);
|
||
mockFindPending.mockResolvedValue(messages);
|
||
|
||
// 启动第一次 poll（未完成）
|
||
const firstPoll = poll();
|
||
|
||
// 尝试第二次 poll（应被跳过）
|
||
await poll();
|
||
|
||
// 只有第一次的 findPending 被调用了一次
|
||
// （第二次 poll 因为 isPolling=true 直接返回）
|
||
expect(mockFindPending).toHaveBeenCalledTimes(1);
|
||
|
||
// 完成
|
||
resolveSend!();
|
||
await firstPoll;
|
||
});
|
||
});
|
||
|
||
// ----------------------------------------------------------
|
||
// dispatch —— 成功路径
|
||
// ----------------------------------------------------------
|
||
describe("dispatch 成功", () => {
|
||
it("应发送到正确的 topic 并标记 published", async () => {
|
||
const message = createMessage({
|
||
eventType: "notification.read",
|
||
aggregateId: "notif-1",
|
||
});
|
||
|
||
await dispatch(message);
|
||
|
||
expect(mockProducer.send).toHaveBeenCalledWith({
|
||
topic: "edu.notification.read",
|
||
messages: [
|
||
{
|
||
key: "notif-1",
|
||
value: JSON.stringify({ notificationId: "n1", userId: "u1" }),
|
||
headers: expect.objectContaining({
|
||
eventId: "evt-1",
|
||
eventType: "notification.read",
|
||
aggregateType: "Notification",
|
||
aggregateId: "notif-1",
|
||
}),
|
||
},
|
||
],
|
||
});
|
||
expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
|
||
});
|
||
|
||
it("payload 为字符串时应直接使用", async () => {
|
||
const message = createMessage({
|
||
payload: "raw-string-payload" as unknown,
|
||
} as OutboxEvent);
|
||
|
||
await dispatch(message);
|
||
|
||
const sendArg = mockProducer.send.mock.calls[0][0];
|
||
expect(sendArg.messages[0].value).toBe("raw-string-payload");
|
||
});
|
||
|
||
it("payload 为对象时应 JSON 序列化", async () => {
|
||
const payload = { key: "value", num: 42 };
|
||
const message = createMessage({ payload });
|
||
|
||
await dispatch(message);
|
||
|
||
const sendArg = mockProducer.send.mock.calls[0][0];
|
||
expect(sendArg.messages[0].value).toBe(JSON.stringify(payload));
|
||
});
|
||
|
||
it("metadata 应合并到 headers", async () => {
|
||
const message = createMessage({
|
||
metadata: { userId: "u1", source: "test" },
|
||
});
|
||
|
||
await dispatch(message);
|
||
|
||
const sendArg = mockProducer.send.mock.calls[0][0];
|
||
expect(sendArg.messages[0].headers).toEqual(
|
||
expect.objectContaining({
|
||
eventId: "evt-1",
|
||
eventType: "notification.sent",
|
||
aggregateType: "Notification",
|
||
aggregateId: "agg-1",
|
||
userId: "u1",
|
||
source: "test",
|
||
}),
|
||
);
|
||
});
|
||
|
||
it("metadata 为 null 时 headers 只含基础字段", async () => {
|
||
const message = createMessage({ metadata: null });
|
||
|
||
await dispatch(message);
|
||
|
||
const sendArg = mockProducer.send.mock.calls[0][0];
|
||
const headers = sendArg.messages[0].headers;
|
||
expect(Object.keys(headers)).toEqual(
|
||
expect.arrayContaining([
|
||
"eventId",
|
||
"eventType",
|
||
"aggregateType",
|
||
"aggregateId",
|
||
]),
|
||
);
|
||
expect(Object.keys(headers)).toHaveLength(4);
|
||
});
|
||
});
|
||
|
||
// ----------------------------------------------------------
|
||
// dispatch —— 重试逻辑
|
||
// ----------------------------------------------------------
|
||
describe("dispatch 重试逻辑", () => {
|
||
it("失败且 retryCount < MAX_RETRY → incrementRetry", async () => {
|
||
const message = createMessage({ retryCount: 2 });
|
||
mockProducer.send.mockRejectedValue(new Error("Kafka timeout"));
|
||
|
||
await dispatch(message);
|
||
|
||
// retryCount=2, +1=3 < MAX_RETRY(5)
|
||
expect(mockIncrementRetry).toHaveBeenCalledTimes(1);
|
||
expect(mockIncrementRetry).toHaveBeenCalledWith(
|
||
"evt-1",
|
||
"Kafka timeout",
|
||
2000 * 2 ** 2, // RETRY_BACKOFF_MS * 2^retryCount = 2000 * 4 = 8000
|
||
);
|
||
expect(mockMarkFailed).not.toHaveBeenCalled();
|
||
});
|
||
|
||
it("失败且 retryCount+1 >= MAX_RETRY → markFailed", async () => {
|
||
const message = createMessage({ retryCount: 4 }); // 4+1=5 >= 5
|
||
mockProducer.send.mockRejectedValue(new Error("Kafka down"));
|
||
|
||
await dispatch(message);
|
||
|
||
expect(mockMarkFailed).toHaveBeenCalledTimes(1);
|
||
expect(mockMarkFailed).toHaveBeenCalledWith("evt-1", "Kafka down");
|
||
expect(mockIncrementRetry).not.toHaveBeenCalled();
|
||
});
|
||
|
||
it("失败且 retryCount=5（超过 MAX_RETRY）→ markFailed", async () => {
|
||
const message = createMessage({ retryCount: 5 });
|
||
mockProducer.send.mockRejectedValue(new Error("Still failing"));
|
||
|
||
await dispatch(message);
|
||
|
||
expect(mockMarkFailed).toHaveBeenCalledWith("evt-1", "Still failing");
|
||
});
|
||
|
||
it("非 Error 类型的异常应转字符串", async () => {
|
||
const message = createMessage({ retryCount: 0 });
|
||
mockProducer.send.mockRejectedValue("string error");
|
||
|
||
await dispatch(message);
|
||
|
||
expect(mockIncrementRetry).toHaveBeenCalledWith(
|
||
"evt-1",
|
||
"string error",
|
||
2000, // 2000 * 2^0 = 2000
|
||
);
|
||
});
|
||
|
||
it("指数退避应正确计算：retryCount=0 → 2000ms, retryCount=1 → 4000ms", async () => {
|
||
mockProducer.send.mockRejectedValue(new Error("fail"));
|
||
|
||
// retryCount=0 → 2000ms
|
||
await dispatch(createMessage({ retryCount: 0 }));
|
||
expect(mockIncrementRetry).toHaveBeenLastCalledWith(
|
||
"evt-1",
|
||
"fail",
|
||
2000,
|
||
);
|
||
|
||
// retryCount=1 → 4000ms
|
||
await dispatch(createMessage({ retryCount: 1 }));
|
||
expect(mockIncrementRetry).toHaveBeenLastCalledWith(
|
||
"evt-1",
|
||
"fail",
|
||
4000,
|
||
);
|
||
});
|
||
});
|
||
|
||
// ----------------------------------------------------------
|
||
// 完整 poll → dispatch → markPublished 流程
|
||
// ----------------------------------------------------------
|
||
describe("完整流程", () => {
|
||
it("poll → dispatch 多条 → 全部 markPublished", async () => {
|
||
const messages = [
|
||
createMessage({ eventId: "evt-1", eventType: "notification.sent" }),
|
||
createMessage({ eventId: "evt-2", eventType: "notification.read" }),
|
||
createMessage({
|
||
eventId: "evt-3",
|
||
eventType: "notification.recalled",
|
||
}),
|
||
];
|
||
mockFindPending.mockResolvedValue(messages);
|
||
|
||
await poll();
|
||
|
||
expect(mockProducer.send).toHaveBeenCalledTimes(3);
|
||
expect(mockMarkPublished).toHaveBeenCalledTimes(3);
|
||
|
||
// 验证每条消息发送到正确 topic
|
||
const topics = mockProducer.send.mock.calls.map(
|
||
(call) => (call[0] as { topic: string }).topic,
|
||
);
|
||
expect(topics).toEqual([
|
||
"edu.notification.sent",
|
||
"edu.notification.read",
|
||
"edu.notification.recalled",
|
||
]);
|
||
});
|
||
|
||
it("poll 中部分 dispatch 失败不应中断后续消息", async () => {
|
||
const messages = [
|
||
createMessage({ eventId: "evt-1" }),
|
||
createMessage({ eventId: "evt-2" }),
|
||
createMessage({ eventId: "evt-3" }),
|
||
];
|
||
mockFindPending.mockResolvedValue(messages);
|
||
// 第二条失败
|
||
mockProducer.send
|
||
.mockResolvedValueOnce({} as never)
|
||
.mockRejectedValueOnce(new Error("Kafka error"))
|
||
.mockResolvedValueOnce({} as never);
|
||
|
||
await poll();
|
||
|
||
// 第一条和第三条成功 markPublished，第二条 incrementRetry
|
||
expect(mockMarkPublished).toHaveBeenCalledTimes(2);
|
||
expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
|
||
expect(mockMarkPublished).toHaveBeenCalledWith("evt-3");
|
||
expect(mockIncrementRetry).toHaveBeenCalledTimes(1);
|
||
expect(mockIncrementRetry).toHaveBeenCalledWith(
|
||
"evt-2",
|
||
"Kafka error",
|
||
2000,
|
||
);
|
||
});
|
||
});
|
||
});
|