diff --git a/packages/shared-ts/src/outbox/index.ts b/packages/shared-ts/src/outbox/index.ts index 41c3eef..2fad7ce 100644 --- a/packages/shared-ts/src/outbox/index.ts +++ b/packages/shared-ts/src/outbox/index.ts @@ -1,20 +1,17 @@ export { OutboxModule } from "./outbox.module.js"; export type { OutboxRootOptions } from "./outbox.module.js"; export { OutboxService } from "./outbox.service.js"; -export { OutboxPublisher } from "./publisher.js"; export { createOutboxTable } from "./schema.js"; export type { OutboxTable, OutboxRow, NewOutboxRow } from "./schema.js"; export { OUTBOX_CONFIG, OUTBOX_DB, - OUTBOX_KAFKA_PRODUCER, OUTBOX_LOGGER, OUTBOX_TABLE, } from "./types.js"; export type { OutboxConfig, OutboxDbClient, - OutboxKafkaProducer, OutboxLogger, OutboxRecord, OutboxStatus, diff --git a/packages/shared-ts/src/outbox/outbox.module.ts b/packages/shared-ts/src/outbox/outbox.module.ts index d3ca866..eb6bab9 100644 --- a/packages/shared-ts/src/outbox/outbox.module.ts +++ b/packages/shared-ts/src/outbox/outbox.module.ts @@ -8,34 +8,29 @@ type PinoFn = (options?: LoggerOptions | unknown) => PinoLogger; const pino = ((pinoNs as unknown as { default: PinoFn }).default ?? (pinoNs as unknown as PinoFn)) as PinoFn; import { OutboxService } from "./outbox.service.js"; -import { OutboxPublisher } from "./publisher.js"; import { createOutboxTable } from "./schema.js"; import { OUTBOX_CONFIG, OUTBOX_DB, - OUTBOX_KAFKA_PRODUCER, OUTBOX_LOGGER, OUTBOX_TABLE, type OutboxConfig, type OutboxDbClient, - type OutboxKafkaProducer, type OutboxLogger, } from "./types.js"; /** * forRoot 选项。 * - * `config` 即 OutboxConfig（表名 / topic / 轮询参数）。 - * `db` / `kafkaProducer` / `logger` 为基础设施实例：NestJS DynamicModule + * `config` 即 OutboxConfig（表名 / topic / maxRetryCount）。 + * `db` / `logger` 为基础设施实例：NestJS DynamicModule * 无法跨模块边界注入宿主已存在的具体单例，故统一通过 forRoot 注入。 * - * `kafkaProducer` 应由调用方以 idempotent 方式创建： - * `kafka.producer({ idempotent: true, transactionalId: '-tx' })`。 + * v2.1（M8）：投递由 Debezium CDC 接管，`kafkaProducer` 参数已移除。 */ export interface OutboxRootOptions { config: OutboxConfig; db: OutboxDbClient; - kafkaProducer: OutboxKafkaProducer; /** 可选；省略时使用默认 pino logger（name=outbox） */ logger?: OutboxLogger; } @@ -47,14 +42,16 @@ function resolveLogger(logger: OutboxLogger | undefined): OutboxLogger { /** * OutboxModule —— 事务性 Outbox 模式的 NestJS DynamicModule。 * - * 注册并启动： + * 注册： * - OutboxService：业务代码注入后调用 `publish` 写入 outbox 记录 - * - OutboxPublisher：onModuleInit 启动轮询，onModuleDestroy 停止轮询 + * + * v2.1（M8）：OutboxPublisher 轮询线程已移除，outbox 表的 binlog 由 + * Debezium 监控并投递到 Kafka（Transaction Log Tailing）。 * * 用法： * ```ts * @Module({ - * imports: [OutboxModule.forRoot({ config, db, kafkaProducer })], + * imports: [OutboxModule.forRoot({ config, db })], * }) * export class AppModule {} * ``` @@ -72,10 +69,8 @@ export class OutboxModule { { provide: OUTBOX_CONFIG, useValue: options.config }, { provide: OUTBOX_DB, useValue: options.db }, { provide: OUTBOX_TABLE, useValue: table }, - { provide: OUTBOX_KAFKA_PRODUCER, useValue: options.kafkaProducer }, { provide: OUTBOX_LOGGER, useValue: logger }, OutboxService, - OutboxPublisher, ], exports: [OutboxService], }; diff --git a/packages/shared-ts/src/outbox/outbox.service.ts b/packages/shared-ts/src/outbox/outbox.service.ts index a171aa7..538b551 100644 --- a/packages/shared-ts/src/outbox/outbox.service.ts +++ b/packages/shared-ts/src/outbox/outbox.service.ts @@ -18,8 +18,9 @@ import { * 调用方在业务事务内调用 `publish`，将事件记录写入 outbox 表， * 与业务写在同一事务中原子提交，保证“业务变更”与“事件发布”的一致性。 * - * 由独立的 OutboxPublisher 负责异步轮询 pending 记录并投递到 Kafka， - * 实现 at-least-once 投递语义。 + * v2.1（M8）：投递由 Debezium CDC 接管——Debezium 监控 MySQL binlog， + * 将 outbox 表的新增记录投递到 Kafka，实现 at-least-once 投递语义。 + * OutboxService 仅负责写入 outbox 表，不再直接调用 Kafka producer。 */ @Injectable() export class OutboxService { diff --git a/packages/shared-ts/src/outbox/publisher.ts b/packages/shared-ts/src/outbox/publisher.ts deleted file mode 100644 index 59ec8e7..0000000 --- a/packages/shared-ts/src/outbox/publisher.ts +++ /dev/null @@ -1,205 +0,0 @@ -import { - Inject, - Injectable, - type OnModuleDestroy, - type OnModuleInit, -} from "@nestjs/common"; -import { and, eq, isNull, lte, or, type SQL } from "drizzle-orm"; -import type { OutboxRow, OutboxTable } from "./schema.js"; -import { - OUTBOX_CONFIG, - OUTBOX_DB, - OUTBOX_KAFKA_PRODUCER, - OUTBOX_LOGGER, - OUTBOX_TABLE, - type OutboxConfig, - type OutboxDbClient, - type OutboxKafkaProducer, - type OutboxLogger, -} from "./types.js"; - -/** - * OutboxPublisher —— 轮询 pending 记录并投递到 Kafka。 - * - * 实现 at-least-once 投递语义： - * 1. 周期性（pollIntervalMs）批量拉取 status=pending 且到期的记录 - * 2. 逐条投递到 Kafka（message key = event id，保证同聚合有序） - * 3. 投递成功 → 标记 published - * 4. 投递失败 → 指数退避更新 nextRetryAt；重试耗尽 → 标记 failed - * - * 幂等性： - * - producer 应由调用方配置为 idempotent（`kafka.producer({ idempotent: true, transactionalId })`）， - * 避免生产端重试产生重复消息 - * - message key = event id（cuid2），消费端可基于 event_id 去重（Redis SETNX 或 DB 唯一索引） - */ -@Injectable() -export class OutboxPublisher implements OnModuleInit, OnModuleDestroy { - private intervalId: ReturnType | null = null; - private isPolling = false; - - constructor( - @Inject(OUTBOX_DB) private readonly db: OutboxDbClient, - @Inject(OUTBOX_TABLE) private readonly table: OutboxTable, - @Inject(OUTBOX_KAFKA_PRODUCER) - private readonly producer: OutboxKafkaProducer, - @Inject(OUTBOX_CONFIG) private readonly config: OutboxConfig, - @Inject(OUTBOX_LOGGER) private readonly logger: OutboxLogger, - ) {} - - async onModuleInit(): Promise { - await this.start(); - } - - async onModuleDestroy(): Promise { - await this.stop(); - } - - async start(): Promise { - if (this.intervalId) return; - this.logger.info( - { - topic: this.config.kafkaTopic, - pollIntervalMs: this.config.pollIntervalMs, - batchSize: this.config.batchSize, - }, - "OutboxPublisher started", - ); - this.intervalId = setInterval(() => { - void this.poll(); - }, this.config.pollIntervalMs); - } - - async stop(): Promise { - if (this.intervalId) { - clearInterval(this.intervalId); - this.intervalId = null; - } - this.logger.info("OutboxPublisher stopped"); - } - - /** - * 拉取一批到期 pending 记录并逐条投递。 - * 通过 isPolling 标志保证不会并发轮询。 - */ - private async poll(): Promise { - if (this.isPolling) return; - this.isPolling = true; - try { - const now = new Date(); - const where: SQL | undefined = and( - eq(this.table.status, "pending"), - or(isNull(this.table.nextRetryAt), lte(this.table.nextRetryAt, now)), - ); - const messages: OutboxRow[] = await this.db - .select() - .from(this.table) - .where(where) - .limit(this.config.batchSize); - - for (const message of messages) { - await this.dispatch(message); - } - } catch (error) { - this.logger.error({ error }, "Outbox poll failed"); - } finally { - this.isPolling = false; - } - } - - /** - * 投递单条记录到 Kafka 并更新状态。 - */ - private async dispatch(message: OutboxRow): Promise { - try { - await this.producer.send({ - topic: this.config.kafkaTopic, - messages: [ - { - key: message.id, - value: message.payload, - headers: this.buildHeaders(message), - }, - ], - }); - - await this.db - .update(this.table) - .set({ status: "published", publishedAt: new Date(), lastError: null }) - .where(eq(this.table.id, message.id)); - - this.logger.info( - { - id: message.id, - eventType: message.eventType, - topic: this.config.kafkaTopic, - }, - "Outbox message published", - ); - } catch (error) { - await this.handleFailure(message, this.toErrorMessage(error)); - } - } - - /** - * 失败处理：指数退避更新 nextRetryAt，重试耗尽则标记 failed。 - */ - private async handleFailure( - message: OutboxRow, - errorMessage: string, - ): Promise { - const newRetryCount = message.retryCount + 1; - this.logger.error( - { - id: message.id, - eventType: message.eventType, - retryCount: newRetryCount, - error: errorMessage, - }, - "Outbox publish failed", - ); - - if (newRetryCount >= message.maxRetryCount) { - await this.db - .update(this.table) - .set({ - status: "failed", - retryCount: newRetryCount, - lastError: errorMessage, - }) - .where(eq(this.table.id, message.id)); - return; - } - - const backoffMs = this.config.retryBackoffMs * 2 ** newRetryCount; - const nextRetryAt = new Date(Date.now() + backoffMs); - await this.db - .update(this.table) - .set({ - retryCount: newRetryCount, - nextRetryAt, - lastError: errorMessage, - }) - .where(eq(this.table.id, message.id)); - } - - /** - * 构建 Kafka 消息头：eventId / eventType / aggregateId + 调用方 metadata。 - */ - private buildHeaders(message: OutboxRow): Record { - const headers: Record = { - eventId: message.id, - eventType: message.eventType, - aggregateId: message.aggregateId, - }; - if (message.metadata) { - for (const [key, value] of Object.entries(message.metadata)) { - headers[key] = value; - } - } - return headers; - } - - private toErrorMessage(error: unknown): string { - return error instanceof Error ? error.message : String(error); - } -} diff --git a/packages/shared-ts/src/outbox/types.ts b/packages/shared-ts/src/outbox/types.ts index b134701..e74d978 100644 --- a/packages/shared-ts/src/outbox/types.ts +++ b/packages/shared-ts/src/outbox/types.ts @@ -1,31 +1,27 @@ import type { MySql2Database } from "drizzle-orm/mysql2"; -import type { Producer } from "kafkajs"; import type { Logger } from "pino"; /** * Outbox 记录状态机： - * - pending：已写入 outbox 表，等待 publisher 投递 - * - published：已成功投递到 Kafka + * - pending：已写入 outbox 表，等待 Debezium 投递 + * - published：已成功投递到 Kafka（由 Debezium/CDC 标记） * - failed：重试次数耗尽，需人工介入 */ export type OutboxStatus = "pending" | "published" | "failed"; /** * Outbox 模块配置（由各服务在 forRoot 时提供）。 + * + * v2.1（M8）：投递由 Debezium CDC 接管，轮询相关配置（pollIntervalMs / + * batchSize / retryBackoffMs）已移除。 */ export interface OutboxConfig { /** outbox 表名（每服务独立，如 core_edu_outbox） */ tableName: string; - /** Kafka 投递目标 topic */ + /** Kafka 投递目标 topic（仅供 Debezium 路由参考，OutboxService 不直接使用） */ kafkaTopic: string; - /** 轮询 pending 记录的间隔（毫秒） */ - pollIntervalMs: number; - /** 每次轮询批量处理的记录数 */ - batchSize: number; - /** 单条记录最大重试次数 */ + /** 单条记录最大重试次数（写入 outbox 行的 max_retry_count 列） */ maxRetryCount: number; - /** 重试退避基数（毫秒），实际退避 = retryBackoffMs * 2^retryCount */ - retryBackoffMs: number; } /** @@ -73,9 +69,6 @@ export interface OutboxRecord { */ export type OutboxDbClient = MySql2Database>; -/** Kafka producer 类型别名 */ -export type OutboxKafkaProducer = Producer; - /** pino Logger 类型别名 */ export type OutboxLogger = Logger; @@ -83,5 +76,4 @@ export type OutboxLogger = Logger; export const OUTBOX_CONFIG = Symbol("OUTBOX_CONFIG"); export const OUTBOX_DB = Symbol("OUTBOX_DB"); export const OUTBOX_TABLE = Symbol("OUTBOX_TABLE"); -export const OUTBOX_KAFKA_PRODUCER = Symbol("OUTBOX_KAFKA_PRODUCER"); export const OUTBOX_LOGGER = Symbol("OUTBOX_LOGGER"); diff --git a/services/content/src/main.ts b/services/content/src/main.ts index b8fc92d..4196a6c 100644 --- a/services/content/src/main.ts +++ b/services/content/src/main.ts @@ -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 { "Content service started (HTTP + gRPC)", ); - // 启动 Outbox Publisher（轮询 pending 事件投递 Kafka） - await outboxPublisher.start(); - // 创建/校验 ES 索引（幂等；ES 不可用时跳过） await ensureQuestionIndex(); @@ -102,7 +98,6 @@ async function bootstrap(): Promise { 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 { logger.info("SIGINT received, shutting down gracefully..."); await esSyncWorker.stop(); await neo4jSyncWorker.stop(); - await outboxPublisher.stop(); await disconnectKafka(); await app.close(); await closeEs(); diff --git a/services/content/src/shared/outbox/outbox.publisher.test.ts b/services/content/src/shared/outbox/outbox.publisher.test.ts deleted file mode 100644 index 0ea983a..0000000 --- a/services/content/src/shared/outbox/outbox.publisher.test.ts +++ /dev/null @@ -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 { - 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 }).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 }).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 }).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 }).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 }).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 }).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(); - }); - }); -}); diff --git a/services/content/src/shared/outbox/outbox.publisher.ts b/services/content/src/shared/outbox/outbox.publisher.ts deleted file mode 100644 index 019aa04..0000000 --- a/services/content/src/shared/outbox/outbox.publisher.ts +++ /dev/null @@ -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 | null = null; - private isPolling = false; - - async start(): Promise { - logger.info("OutboxPublisher started"); - this.intervalId = setInterval(() => { - void this.poll(); - }, POLL_INTERVAL_MS); - } - - async stop(): Promise { - if (this.intervalId) { - clearInterval(this.intervalId); - this.intervalId = null; - } - logger.info("OutboxPublisher stopped"); - } - - private async poll(): Promise { - 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 { - 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(); diff --git a/services/core-edu/src/main.ts b/services/core-edu/src/main.ts index 23e8944..87a52f7 100644 --- a/services/core-edu/src/main.ts +++ b/services/core-edu/src/main.ts @@ -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 { 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 { // 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 { const shutdown = async (signal: string): Promise => { logger.info({ signal }, "Shutting down gracefully..."); await stopGrpcServer(); - await outboxPublisher.stop(); await disconnectRedis(); await disconnectKafka(); await shutdownTracer(); diff --git a/services/core-edu/src/shared/lifecycle/lifecycle.service.ts b/services/core-edu/src/shared/lifecycle/lifecycle.service.ts index bf12f20..3d86c5a 100644 --- a/services/core-edu/src/shared/lifecycle/lifecycle.service.ts +++ b/services/core-edu/src/shared/lifecycle/lifecycle.service.ts @@ -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(); diff --git a/services/core-edu/src/shared/outbox/outbox.publisher.ts b/services/core-edu/src/shared/outbox/outbox.publisher.ts deleted file mode 100644 index 32fa5c7..0000000 --- a/services/core-edu/src/shared/outbox/outbox.publisher.ts +++ /dev/null @@ -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.. - */ -const TOPIC_MAP: Record = { - // 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 { - logger.info("OutboxPublisher started"); - this.intervalId = setInterval(() => { - void this.poll(); - }, POLL_INTERVAL_MS); - } - - async stop(): Promise { - if (this.intervalId) { - clearInterval(this.intervalId); - this.intervalId = null; - } - logger.info("OutboxPublisher stopped"); - } - - private async poll(): Promise { - 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 { - 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(); diff --git a/services/iam/src/app.module.ts b/services/iam/src/app.module.ts index b7d5439..404e74d 100644 --- a/services/iam/src/app.module.ts +++ b/services/iam/src/app.module.ts @@ -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 { - // 连接 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"); diff --git a/services/msg/src/shared/lifecycle/lifecycle.service.ts b/services/msg/src/shared/lifecycle/lifecycle.service.ts index 671ab18..879ee43 100644 --- a/services/msg/src/shared/lifecycle/lifecycle.service.ts +++ b/services/msg/src/shared/lifecycle/lifecycle.service.ts @@ -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 { - // 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()); diff --git a/services/msg/src/shared/outbox/outbox.publisher.ts b/services/msg/src/shared/outbox/outbox.publisher.ts deleted file mode 100644 index 5c47dee..0000000 --- a/services/msg/src/shared/outbox/outbox.publisher.ts +++ /dev/null @@ -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 | null = null; - private isPolling = false; - - async start(): Promise { - 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 { - if (this.intervalId) { - clearInterval(this.intervalId); - this.intervalId = null; - } - logger.info("OutboxPublisher stopped"); - } - - private async poll(): Promise { - 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 { - 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 { - const headers: Record = { - 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(); diff --git a/services/msg/src/shared/outbox/outbox.service.ts b/services/msg/src/shared/outbox/outbox.service.ts index 0324d9f..1702f93 100644 --- a/services/msg/src/shared/outbox/outbox.service.ts +++ b/services/msg/src/shared/outbox/outbox.service.ts @@ -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 实现幂等去重。 */ diff --git a/services/msg/test/unit/outbox.publisher.spec.ts b/services/msg/test/unit/outbox.publisher.spec.ts deleted file mode 100644 index 6dcabfe..0000000 --- a/services/msg/test/unit/outbox.publisher.spec.ts +++ /dev/null @@ -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 = { - "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 { - 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 { - return (outboxPublisher as unknown as { poll: () => Promise }).poll(); -} - -function dispatch(message: OutboxEvent): Promise { - return ( - outboxPublisher as unknown as { - dispatch: (m: OutboxEvent) => Promise; - } - ).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, - ); - }); - }); -});