xiner/Edu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(shared-ts,iam,core-edu,content,msg): remove outbox polling publisher

Browse Source 
M8: Debezium CDC now handles outbox table to Kafka (ADR-032)

- Remove OutboxPublisher class from shared-ts

- Remove publisher from iam/core-edu/content/msg lifecycle and modules

- OutboxService retained for transactional outbox table writes

- Debezium monitors binlog and pushes to Kafka automatically
This commit is contained in:
SpecialX
2026-07-15 01:27:45 +08:00
parent a3f4fd013e
commit 47a062606f
16 changed files with 34 additions and 1227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
services/content/src/main.ts
View File

@@ -12,7 +12,6 @@ import { closeNeo4j } from "./config/neo4j.js";
import { connectKafka, disconnectKafka } from "./config/kafka.js";
import { logger } from "./shared/observability/logger.js";
import { metricsRegistry } from "./shared/observability/metrics.js";
import { outboxPublisher } from "./shared/outbox/outbox.publisher.js";
import { neo4jSyncWorker } from "./shared/sync/neo4j-sync.worker.js";
import { esSyncWorker } from "./shared/sync/es-sync.worker.js";
import { ensureQuestionIndex, closeEs } from "./config/elasticsearch.js";
@@ -86,9 +85,6 @@ async function bootstrap(): Promise<void> {
"Content service started (HTTP + gRPC)",
);
// 启动 Outbox Publisher（轮询 pending 事件投递 Kafka）
await outboxPublisher.start();
// 创建/校验 ES 索引（幂等；ES 不可用时跳过）
await ensureQuestionIndex();
@@ -102,7 +98,6 @@ async function bootstrap(): Promise<void> {
logger.info("SIGTERM received, shutting down gracefully...");
await esSyncWorker.stop();
await neo4jSyncWorker.stop();
await outboxPublisher.stop();
await disconnectKafka();
await app.close();
await closeEs();
@@ -116,7 +111,6 @@ async function bootstrap(): Promise<void> {
logger.info("SIGINT received, shutting down gracefully...");
await esSyncWorker.stop();
await neo4jSyncWorker.stop();
await outboxPublisher.stop();
await disconnectKafka();
await app.close();
await closeEs();

163
services/content/src/shared/outbox/outbox.publisher.test.ts
View File

@@ -1,163 +0,0 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
const mockProducer = vi.hoisted(() => ({ send: vi.fn() }));
vi.mock("../../config/kafka.js", () => ({
producer: mockProducer,
}));
vi.mock("../observability/logger.js", () => ({
logger: {
info: vi.fn(),
error: vi.fn(),
warn: vi.fn(),
debug: vi.fn(),
},
}));
vi.mock("./outbox.repository.js", () => ({
outboxRepository: {
findPending: vi.fn(),
markPublished: vi.fn(),
incrementRetry: vi.fn(),
markFailed: vi.fn(),
create: vi.fn(),
},
}));
import { OutboxPublisher } from "./outbox.publisher.js";
import { outboxRepository } from "./outbox.repository.js";
import type { OutboxMessage } from "./outbox.schema.js";
function createMessage(overrides: Partial<OutboxMessage> = {}): OutboxMessage {
return {
id: "msg-1",
aggregateType: "Chapter",
aggregateId: "ch-1",
eventType: "chapter.created",
topic: "edu.content.chapter.events",
payload: '{"event_id":"e1"}',
status: "pending",
retryCount: 0,
createdAt: new Date(),
publishedAt: null,
nextRetryAt: null,
lastError: null,
...overrides,
};
}
describe("OutboxPublisher", () => {
let publisher: OutboxPublisher;
beforeEach(() => {
vi.clearAllMocks();
publisher = new OutboxPublisher();
});
describe("dispatch (via poll)", () => {
it("should send message to kafka and mark as published", async () => {
const message = createMessage();
vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
mockProducer.send.mockResolvedValue(undefined);
// Access private poll via casting
await (publisher as unknown as { poll: () => Promise<void> }).poll();
expect(mockProducer.send).toHaveBeenCalledWith({
topic: message.topic,
messages: [
{
key: message.aggregateId,
value: message.payload,
headers: {
eventId: message.id,
eventType: message.eventType,
aggregateType: message.aggregateType,
},
},
],
});
expect(outboxRepository.markPublished).toHaveBeenCalledWith(message.id);
});
it("should increment retry when send fails and below max retries", async () => {
const message = createMessage({ id: "msg-2", retryCount: 0 });
vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
mockProducer.send.mockRejectedValue(new Error("kafka down"));
await (publisher as unknown as { poll: () => Promise<void> }).poll();
expect(outboxRepository.incrementRetry).toHaveBeenCalledWith(
"msg-2",
"kafka down",
);
expect(outboxRepository.markFailed).not.toHaveBeenCalled();
});
it("should mark failed when retry count reaches max", async () => {
const message = createMessage({ id: "msg-3", retryCount: 4 });
vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
mockProducer.send.mockRejectedValue(new Error("kafka down"));
await (publisher as unknown as { poll: () => Promise<void> }).poll();
expect(outboxRepository.markFailed).toHaveBeenCalledWith(
"msg-3",
"kafka down",
);
expect(outboxRepository.incrementRetry).not.toHaveBeenCalled();
});
it("should handle non-Error rejection in dispatch", async () => {
const message = createMessage({ id: "msg-4", retryCount: 0 });
vi.mocked(outboxRepository.findPending).mockResolvedValue([message]);
mockProducer.send.mockRejectedValue("string error");
await (publisher as unknown as { poll: () => Promise<void> }).poll();
expect(outboxRepository.incrementRetry).toHaveBeenCalledWith(
"msg-4",
"string error",
);
});
});
describe("poll error handling", () => {
it("should swallow errors from findPending", async () => {
vi.mocked(outboxRepository.findPending).mockRejectedValue(
new Error("db error"),
);
await expect(
(publisher as unknown as { poll: () => Promise<void> }).poll(),
).resolves.not.toThrow();
});
it("should process multiple messages in a batch", async () => {
const messages = [
createMessage({ id: "m1" }),
createMessage({ id: "m2" }),
];
vi.mocked(outboxRepository.findPending).mockResolvedValue(messages);
mockProducer.send.mockResolvedValue(undefined);
await (publisher as unknown as { poll: () => Promise<void> }).poll();
expect(mockProducer.send).toHaveBeenCalledTimes(2);
expect(outboxRepository.markPublished).toHaveBeenCalledWith("m1");
expect(outboxRepository.markPublished).toHaveBeenCalledWith("m2");
});
});
describe("start/stop", () => {
it("should start and stop without error", async () => {
await publisher.start();
await publisher.stop();
// Should not throw
});
it("should stop without error when not started", async () => {
await publisher.stop();
});
});
});

81
services/content/src/shared/outbox/outbox.publisher.ts
View File

@@ -1,81 +0,0 @@
import { logger } from "../observability/logger.js";
import { producer } from "../../config/kafka.js";
import { outboxRepository } from "./outbox.repository.js";
import type { OutboxMessage } from "./outbox.schema.js";
const POLL_INTERVAL_MS = 5000;
const BATCH_SIZE = 100;
const MAX_RETRY = 5;
export class OutboxPublisher {
private intervalId: ReturnType<typeof setInterval> | null = null;
private isPolling = false;
async start(): Promise<void> {
logger.info("OutboxPublisher started");
this.intervalId = setInterval(() => {
void this.poll();
}, POLL_INTERVAL_MS);
}
async stop(): Promise<void> {
if (this.intervalId) {
clearInterval(this.intervalId);
this.intervalId = null;
}
logger.info("OutboxPublisher stopped");
}
private async poll(): Promise<void> {
if (this.isPolling) return;
this.isPolling = true;
try {
const messages = await outboxRepository.findPending(BATCH_SIZE);
for (const message of messages) {
await this.dispatch(message);
}
} catch (error) {
logger.error({ error }, "Outbox poll failed");
} finally {
this.isPolling = false;
}
}
private async dispatch(message: OutboxMessage): Promise<void> {
try {
await producer.send({
topic: message.topic,
messages: [
{
key: message.aggregateId,
value: message.payload,
headers: {
eventId: message.id,
eventType: message.eventType,
aggregateType: message.aggregateType,
},
},
],
});
await outboxRepository.markPublished(message.id);
logger.info(
{ id: message.id, eventType: message.eventType, topic: message.topic },
"Outbox message published",
);
} catch (error) {
const errorMessage =
error instanceof Error ? error.message : String(error);
logger.error(
{ error, id: message.id, eventType: message.eventType },
"Outbox publish failed",
);
if (message.retryCount + 1 >= MAX_RETRY) {
await outboxRepository.markFailed(message.id, errorMessage);
} else {
await outboxRepository.incrementRetry(message.id, errorMessage);
}
}
}
}
export const outboxPublisher = new OutboxPublisher();

12
services/core-edu/src/main.ts
View File

@@ -3,7 +3,6 @@ import { AppModule } from "./app.module.js";
import { env } from "./config/env.js";
import { connectKafka, disconnectKafka } from "./config/kafka.js";
import { connectRedis, disconnectRedis } from "./config/redis.js";
import { outboxPublisher } from "./shared/outbox/outbox.publisher.js";
import { GlobalErrorFilter } from "./shared/errors/global-error.filter.js";
import { initTracer, shutdownTracer } from "./shared/observability/tracer.js";
import { logger } from "./shared/observability/logger.js";
@@ -24,9 +23,9 @@ async function bootstrap(): Promise<void> {
res.end(await registry.metrics());
});
// Connect Kafka producer/consumer before starting the outbox publisher.
// Non-blocking: if Kafka is unavailable, service still starts; outbox
// publisher will retry sends and messages stay pending until Kafka recovers.
// Connect Kafka producer/consumer.
// Non-blocking: if Kafka is unavailable, service still starts.
// v2.1（M8）：OutboxPublisher 轮询线程已移除，outbox 投递由 Debezium CDC 接管。
void connectKafka();
// Connect Redis for distributed lock (homework submission idempotency).
@@ -34,10 +33,6 @@ async function bootstrap(): Promise<void> {
// operations will fall back to DB unique index for idempotency.
void connectRedis();
// Start the transactional outbox publisher - polls pending messages
// and publishes them to Kafka topics defined in TOPIC_MAP.
await outboxPublisher.start();
await app.listen(env.PORT);
// 启动 gRPC server（9 Service / 40 RPC）
@@ -51,7 +46,6 @@ async function bootstrap(): Promise<void> {
const shutdown = async (signal: string): Promise<void> => {
logger.info({ signal }, "Shutting down gracefully...");
await stopGrpcServer();
await outboxPublisher.stop();
await disconnectRedis();
await disconnectKafka();
await shutdownTracer();

2
services/core-edu/src/shared/lifecycle/lifecycle.service.ts
View File

@@ -7,7 +7,6 @@ import {
import { closeDb } from "../../config/database.js";
import { disconnectRedis } from "../../config/redis.js";
import { disconnectKafka } from "../../config/kafka.js";
import { outboxPublisher } from "../outbox/outbox.publisher.js";
const SERVICE_NAME = "core-edu";
@@ -24,7 +23,6 @@ export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
`service ${SERVICE_NAME} shutting down (signal=${signal ?? "unknown"})`,
);
try {
await outboxPublisher.stop();
await disconnectRedis();
await disconnectKafka();
await closeDb();

119
services/core-edu/src/shared/outbox/outbox.publisher.ts
View File

@@ -1,119 +0,0 @@
import { logger } from "../observability/logger.js";
import { producer } from "../../config/kafka.js";
import { outboxRepository } from "./outbox.repository.js";
import type { OutboxMessage } from "./outbox.schema.js";
/**
* TOPIC_MAP: 事件类型 → Kafka topic
*
* 仲裁依据：ISSUE-002 / ISSUE-004 / events.proto 头注释
* 命名规范：edu.teaching.<aggregate>.<action>
*/
const TOPIC_MAP: Record<string, string> = {
// Exam events
"exam.created": "edu.teaching.exam.created",
"exam.updated": "edu.teaching.exam.updated",
"exam.published": "edu.teaching.exam.published",
"exam.submitted": "edu.teaching.exam.submitted",
"exam.graded": "edu.teaching.exam.graded",
"exam.deleted": "edu.teaching.exam.deleted",
// Exam realtime events (P3.14: 供 msg → push-gateway → student-portal WebSocket)
"exam.extended": "edu.teaching.exam.extended",
"exam.force_submitted": "edu.teaching.exam.force_submitted",
"exam.question_reordered": "edu.teaching.exam.question_reordered",
// Homework events
"homework.assigned": "edu.teaching.homework.assigned",
"homework.submitted": "edu.teaching.homework.submitted",
"homework.graded": "edu.teaching.homework.graded",
// Grade events
"grade.recorded": "edu.teaching.grade.recorded",
"grade.updated": "edu.teaching.grade.updated",
// Attendance events
"attendance.recorded": "edu.teaching.attendance.recorded",
// Class events (ISSUE-004: 统一为 edu.teaching.class.transferred)
"class.transferred": "edu.teaching.class.transferred",
};
const POLL_INTERVAL_MS = 5000;
const BATCH_SIZE = 100;
const MAX_RETRY = 5;
const RETRY_BACKOFF_BASE_MS = 1000;
export class OutboxPublisher {
private intervalId: NodeJS.Timeout | null = null;
private isPolling = false;
async start(): Promise<void> {
logger.info("OutboxPublisher started");
this.intervalId = setInterval(() => {
void this.poll();
}, POLL_INTERVAL_MS);
}
async stop(): Promise<void> {
if (this.intervalId) {
clearInterval(this.intervalId);
this.intervalId = null;
}
logger.info("OutboxPublisher stopped");
}
private async poll(): Promise<void> {
if (this.isPolling) return;
this.isPolling = true;
try {
const messages = await outboxRepository.findPending(BATCH_SIZE);
for (const message of messages) {
await this.publish(message);
}
} catch (error) {
logger.error({ error }, "Outbox poll failed");
} finally {
this.isPolling = false;
}
}
private async publish(message: OutboxMessage): Promise<void> {
const topic = TOPIC_MAP[message.eventType] ?? "edu.teaching.fallback";
try {
await producer.send({
topic,
messages: [
{
key: message.aggregateId,
value: message.payload,
headers: {
eventType: message.eventType,
aggregateType: message.aggregateType,
eventId: message.eventId,
},
},
],
});
await outboxRepository.markProcessed(message.id);
logger.info(
{
id: message.id,
eventId: message.eventId,
eventType: message.eventType,
topic,
},
"Outbox message published",
);
} catch (error) {
logger.error(
{ error, id: message.id, eventId: message.eventId },
"Outbox publish failed",
);
const nextRetry = message.retryCount + 1;
if (nextRetry >= MAX_RETRY) {
await outboxRepository.markFailed(message.id);
} else {
const backoff = RETRY_BACKOFF_BASE_MS * Math.pow(2, nextRetry);
await outboxRepository.incrementRetry(message.id, backoff);
}
}
}
}
export const outboxPublisher = new OutboxPublisher();

15
services/iam/src/app.module.ts
View File

@@ -15,11 +15,7 @@ import { GraphqlModule } from "./graphql/graphql.module.js";
import { RouterAuthGuard } from "./graphql/router-auth.guard.js";
import { OutboxModule } from "@edu/shared-ts/outbox";
import { getDbInstance } from "./config/database.js";
import {
getKafkaProducer,
connectKafkaProducer,
IAM_KAFKA_TOPICS,
} from "./config/kafka.js";
import { connectKafkaProducer, IAM_KAFKA_TOPICS } from "./config/kafka.js";
/**
* IAM 根模块（v2.1）。
@@ -42,13 +38,9 @@ import {
config: {
tableName: "iam_outbox",
kafkaTopic: IAM_KAFKA_TOPICS.USER_EVENTS,
pollIntervalMs: 1000,
batchSize: 20,
maxRetryCount: 5,
retryBackoffMs: 1000,
},
db: getDbInstance(),
kafkaProducer: getKafkaProducer(),
}),
],
providers: [
@@ -61,9 +53,8 @@ export class AppModule implements NestModule, OnModuleInit {
private readonly logger = new Logger(AppModule.name);
async onModuleInit(): Promise<void> {
// 连接 Kafka producer（Outbox 投递前置依赖）
// v2.1 注：OutboxPublisher 轮询线程将被废弃（M8），由 Debezium 接管投递
// 此处保留 Kafka producer 连接用于其他场景（如直接发事件）
// 连接 Kafka producer（健康检查 /healthz 依赖 producer 探活）
// v2.1（M8）：OutboxPublisher 轮询线程已移除，outbox 投递由 Debezium CDC 接管
try {
await connectKafkaProducer();
this.logger.log("Kafka producer connected");

21
services/msg/src/shared/lifecycle/lifecycle.service.ts
View File

@@ -8,25 +8,24 @@ import { closeDb } from "../../config/database.js";
import { closeEs } from "../../config/elasticsearch.js";
import { connectKafka, disconnectKafka } from "../kafka/kafka.client.js";
import { closeRedis } from "../redis/redis.client.js";
import { outboxPublisher } from "../outbox/outbox.publisher.js";
const SERVICE_NAME = "msg";
/**
* 优雅停机服务 + 资源生命周期管理。
*
* 仲裁依据 G2：统一管理 DB/ES/Redis/Kafka/OutboxPublisher 生命周期。
* 仲裁依据 G2：统一管理 DB/ES/Redis/Kafka 生命周期。
*
* v2.1（M8）：OutboxPublisher 轮询线程已移除，outbox 投递由 Debezium CDC 接管。
*
* 启动顺序（onModuleInit）：
* 1. connectKafka()：连接 producer + consumer（幂等，KafkaConsumerService 也会调用）
* 2. outboxPublisher.start()：启动轮询 worker，投递 pending 事件到 Kafka
*
* 关闭顺序（onApplicationShutdown，在 OnModuleDestroy 之后执行）：
* 1. outboxPublisher.stop()：停止轮询
* 2. disconnectKafka()：断开 producer + consumer
* 3. closeRedis()：关闭 Redis 连接
* 4. closeEs()：关闭 ES 客户端
* 5. closeDb()：关闭 MySQL 连接池
* 1. disconnectKafka()：断开 producer + consumer
* 2. closeRedis()：关闭 Redis 连接
* 3. closeEs()：关闭 ES 客户端
* 4. closeDb()：关闭 MySQL 连接池
*
* 注：KafkaConsumerService 实现 OnModuleDestroy，其 stop() 会在
* onApplicationShutdown 之前被 NestJS 调用。disconnectKafka() 是幂等的。
@@ -36,12 +35,9 @@ export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
async onModuleInit(): Promise<void> {
// 1. 连接 Kafka（producer + consumer）
// 连接 Kafka（producer + consumer）
await connectKafka();
// 2. 启动 Outbox publisher 轮询
await outboxPublisher.start();
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
@@ -51,7 +47,6 @@ export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
);
// 按依赖反序关闭
await this.safeStop("outboxPublisher", () => outboxPublisher.stop());
await this.safeStop("kafka", () => disconnectKafka());
await this.safeStop("redis", () => closeRedis());
await this.safeStop("elasticsearch", () => closeEs());

125
services/msg/src/shared/outbox/outbox.publisher.ts
View File

@@ -1,125 +0,0 @@
import { logger } from "../observability/logger.js";
import { getProducer } from "../kafka/kafka.client.js";
import { resolveTopic } from "../kafka/topic-map.js";
import {
findPending,
incrementRetry,
markFailed,
markPublished,
} from "./outbox.repository.js";
import type { OutboxEvent } from "./outbox.schema.js";
/**
* OutboxPublisher —— 轮询 pending 记录并投递到 Kafka（多 topic 路由）。
*
* 参照 core-edu OutboxPublisher 模式，支持 TOPIC_MAP 多 topic 路由：
* - notification.sent → edu.notify.notification.sent
* - notification.read → edu.notify.notification.read
* - notification.recalled → edu.notify.notification.recalled
* - notification.failed → edu.notify.notification.failed
*
* 仲裁依据：at-least-once 投递 + 指数退避重试 + 幂等（消费端 event_id 去重）。
*/
const POLL_INTERVAL_MS = 5000;
const BATCH_SIZE = 100;
const MAX_RETRY = 5;
const RETRY_BACKOFF_MS = 2000;
class OutboxPublisher {
private intervalId: ReturnType<typeof setInterval> | null = null;
private isPolling = false;
async start(): Promise<void> {
if (this.intervalId) return;
logger.info(
{ pollIntervalMs: POLL_INTERVAL_MS, batchSize: BATCH_SIZE },
"OutboxPublisher started",
);
this.intervalId = setInterval(() => {
void this.poll();
}, POLL_INTERVAL_MS);
}
async stop(): Promise<void> {
if (this.intervalId) {
clearInterval(this.intervalId);
this.intervalId = null;
}
logger.info("OutboxPublisher stopped");
}
private async poll(): Promise<void> {
if (this.isPolling) return;
this.isPolling = true;
try {
const messages = await findPending(BATCH_SIZE);
for (const message of messages) {
await this.dispatch(message);
}
} catch (error) {
logger.error({ error }, "Outbox poll failed");
} finally {
this.isPolling = false;
}
}
private async dispatch(message: OutboxEvent): Promise<void> {
const topic = resolveTopic(message.eventType);
try {
const producer = getProducer();
await producer.send({
topic,
messages: [
{
key: message.aggregateId,
value:
typeof message.payload === "string"
? message.payload
: JSON.stringify(message.payload),
headers: this.buildHeaders(message),
},
],
});
await markPublished(message.eventId);
logger.info(
{ eventId: message.eventId, eventType: message.eventType, topic },
"Outbox message published",
);
} catch (error) {
const errorMessage =
error instanceof Error ? error.message : String(error);
logger.error(
{
eventId: message.eventId,
eventType: message.eventType,
error: errorMessage,
},
"Outbox publish failed",
);
if (message.retryCount + 1 >= MAX_RETRY) {
await markFailed(message.eventId, errorMessage);
} else {
const backoffMs = RETRY_BACKOFF_MS * 2 ** message.retryCount;
await incrementRetry(message.eventId, errorMessage, backoffMs);
}
}
}
private buildHeaders(message: OutboxEvent): Record<string, string> {
const headers: Record<string, string> = {
eventId: message.eventId,
eventType: message.eventType,
aggregateType: message.aggregateType,
aggregateId: message.aggregateId,
};
if (message.metadata) {
for (const [key, value] of Object.entries(message.metadata)) {
headers[key] = value;
}
}
return headers;
}
}
export const outboxPublisher = new OutboxPublisher();

3
services/msg/src/shared/outbox/outbox.service.ts
View File

@@ -10,7 +10,8 @@ import { outboxEvents } from "./outbox.schema.js";
* 调用方在业务事务内调用 publish()，将事件记录写入 msg_outbox_events 表，
* 与业务写在同一事务中原子提交（事务性 Outbox 模式）。
*
* 由 OutboxPublisher 负责异步轮询 pending 记录并投递到 Kafka（at-least-once）。
* v2.1（M8）：投递由 Debezium CDC 接管——监控 MySQL binlog 将 outbox 记录
* 投递到 Kafka（at-least-once）。OutboxService 仅负责写入 outbox 表。
*
* 仲裁依据 G11：eventId 用 cuid2，同时作为 Kafka 消息 key 实现幂等去重。
*/

458
services/msg/test/unit/outbox.publisher.spec.ts
View File

@@ -1,458 +0,0 @@
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,
);
});
});
});