feat(msg): v2 ARB-013 topic 命名统一 + 考试事件消费

ARB-013 P0 修复：PRODUCER_TOPIC_MAP 从 edu.notification.* 改为 edu.notify.notification.* kafka.consumer 新增 3 考试实时事件消费（exam.extended/force_submitted/question_reordered）嵌套 payload 解包支持 + topic-map 扩展新增 6 测试数据文件（docker-notify + 5 kafka 事件 json） 101 单元测试通过 + Docker 真实环境验证
2026-07-14 23:00:33 +08:00
parent ad39a3bb0f
commit 765f7da4c0
13 changed files with 695 additions and 35 deletions
--- a/services/msg/src/shared/kafka/kafka.consumer.ts
+++ b/services/msg/src/shared/kafka/kafka.consumer.ts
@@ -21,7 +21,7 @@ import { NotificationsService } from "../../notifications/notifications.service.
 *
 * 事件路由：
 * - iam：user.created/updated/deleted/role_changed, role.created/updated
- * - core-edu：exam.published, assignment.submitted/graded, grade.recorded, attendance.recorded
+ * - core-edu：exam.published, homework.assigned, assignment.submitted/graded, grade.recorded, attendance.recorded
 * - data-ana：mastery.updated
 */
@Injectable()
@@ -153,22 +153,34 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {

  /**
   * 根据 topic 路由到具体的事件处理器。
+   *
+   * payload 兼容两种格式：
+   * - 扁平格式（测试用）：消息 value 直接是业务字段 { examId, studentIds, ... }
+   * - 嵌套格式（core-edu Outbox 标准）：{ event_id, event_type, payload: { examId, ... } }
+   * routeEvent 内部统一解包到 businessPayload，handler 只处理业务字段。
   */
  private async routeEvent(
    topic: string,
    eventId: string,
    payload: Record<string, unknown>,
  ): Promise<void> {
+    // 解包嵌套 payload：core-edu Outbox 发送 { event_id, event_type, payload: {...} }
+    const inner = payload.payload;
+    const businessPayload =
+      inner && typeof inner === "object" && !Array.isArray(inner)
+        ? (inner as Record<string, unknown>)
+        : payload;
+
    switch (topic) {
      // iam 事件
      case "edu.identity.user.created":
-        await this.handleUserCreated(eventId, payload);
+        await this.handleUserCreated(eventId, businessPayload);
        break;
      case "edu.identity.user.role_changed":
-        await this.handleRoleChanged(eventId, payload);
+        await this.handleRoleChanged(eventId, businessPayload);
        break;
      case "edu.identity.role.updated":
-        await this.handleRoleUpdated(eventId, payload);
+        await this.handleRoleUpdated(eventId, businessPayload);
        break;
      case "edu.identity.user.updated":
      case "edu.identity.user.deleted":
@@ -180,26 +192,44 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
        );
        break;

-      // core-edu 事件
+      // core-edu 事件（6 基础）
      case "edu.teaching.exam.published":
-        await this.handleExamPublished(eventId, payload);
+        await this.handleExamPublished(eventId, businessPayload);
+        break;
+      case "edu.teaching.homework.assigned":
+        await this.handleHomeworkAssigned(eventId, businessPayload);
        break;
      case "edu.teaching.assignment.submitted":
-        await this.handleAssignmentSubmitted(eventId, payload);
+        await this.handleAssignmentSubmitted(eventId, businessPayload);
        break;
      case "edu.teaching.assignment.graded":
-        await this.handleAssignmentGraded(eventId, payload);
+        await this.handleAssignmentGraded(eventId, businessPayload);
        break;
      case "edu.teaching.grade.recorded":
-        await this.handleGradeRecorded(eventId, payload);
+        await this.handleGradeRecorded(eventId, businessPayload);
        break;
      case "edu.teaching.attendance.recorded":
-        await this.handleAttendanceRecorded(eventId, payload);
+        await this.handleAttendanceRecorded(eventId, businessPayload);
+        break;
+
+      // core-edu 考试实时事件（P3.14 新增，3 个）— 转发到 push-gateway
+      case "edu.teaching.exam.extended":
+        await this.handleExamExtended(eventId, businessPayload, payload);
+        break;
+      case "edu.teaching.exam.force_submitted":
+        await this.handleExamForceSubmitted(eventId, businessPayload, payload);
+        break;
+      case "edu.teaching.exam.question_reordered":
+        await this.handleExamQuestionReordered(
+          eventId,
+          businessPayload,
+          payload,
+        );
        break;

      // data-ana 事件
      case "edu.insight.mastery.updated":
-        await this.handleMasteryUpdated(eventId, payload);
+        await this.handleMasteryUpdated(eventId, businessPayload);
        break;

      default:
@@ -207,6 +237,145 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
    }
  }

+  // ============================================================
+  // 考试实时事件处理器（P3.14：转发 core-edu → push-gateway → WebSocket）
+  // ============================================================
+
+  /**
+   * 考试时间延长通知。
+   * payload: { examId, classId, subjectId, extensionSeconds, newDuration, studentIds? }
+   */
+  private async handleExamExtended(
+    eventId: string,
+    payload: Record<string, unknown>,
+    rawPayload: Record<string, unknown>,
+  ): Promise<void> {
+    const examId = String(payload.examId ?? payload.exam_id ?? "");
+    const className = String(payload.className ?? payload.class_name ?? "");
+    const extensionSeconds = Number(
+      payload.extensionSeconds ?? payload.extension_seconds ?? 0,
+    );
+    const newDuration = Number(
+      payload.newDuration ?? payload.new_duration ?? 0,
+    );
+    const studentIds = payload.studentIds ?? payload.student_ids;
+
+    const extensionMin = Math.round(extensionSeconds / 60);
+    const newDurationMin = Math.round(newDuration / 60);
+    const content = `你的考试时间已延长 ${extensionMin} 分钟，新时长：${newDurationMin} 分钟${className ? `（${className}）` : ""}`;
+
+    await this.broadcastExamRealtimeEvent(
+      eventId,
+      "exam",
+      "考试时间延长通知",
+      content,
+      "exam_extended",
+      examId,
+      studentIds,
+      rawPayload,
+    );
+  }
+
+  /**
+   * 考试强制提交通知。
+   * payload: { examId, classId, studentIds? }
+   */
+  private async handleExamForceSubmitted(
+    eventId: string,
+    payload: Record<string, unknown>,
+    rawPayload: Record<string, unknown>,
+  ): Promise<void> {
+    const examId = String(payload.examId ?? payload.exam_id ?? "");
+    const examTitle = String(payload.examTitle ?? payload.exam_title ?? "考试");
+    const studentIds = payload.studentIds ?? payload.student_ids;
+
+    const content = `你的考试「${examTitle}」已被教师强制提交`;
+
+    await this.broadcastExamRealtimeEvent(
+      eventId,
+      "exam",
+      "考试强制提交通知",
+      content,
+      "exam_force_submitted",
+      examId,
+      studentIds,
+      rawPayload,
+    );
+  }
+
+  /**
+   * 考试题目顺序调整通知。
+   * payload: { examId, classId, studentIds? }
+   */
+  private async handleExamQuestionReordered(
+    eventId: string,
+    payload: Record<string, unknown>,
+    rawPayload: Record<string, unknown>,
+  ): Promise<void> {
+    const examId = String(payload.examId ?? payload.exam_id ?? "");
+    const examTitle = String(payload.examTitle ?? payload.exam_title ?? "考试");
+    const studentIds = payload.studentIds ?? payload.student_ids;
+
+    const content = `你的考试「${examTitle}」题目顺序已调整，请刷新查看`;
+
+    await this.broadcastExamRealtimeEvent(
+      eventId,
+      "exam",
+      "题目顺序调整通知",
+      content,
+      "exam_question_reordered",
+      examId,
+      studentIds,
+      rawPayload,
+    );
+  }
+
+  /**
+   * 考试实时事件广播：为每个学生创建 in_app 通知（触发 ChannelDispatcher → push-gateway HTTP + Outbox → Kafka）。
+   * 若 payload 无 studentIds，记录 warn 并跳过（等待 core-edu 补全 payload）。
+   */
+  private async broadcastExamRealtimeEvent(
+    eventId: string,
+    type: string,
+    title: string,
+    content: string,
+    eventTag: string,
+    examId: string,
+    studentIds: unknown,
+    rawPayload: Record<string, unknown>,
+  ): Promise<void> {
+    if (!Array.isArray(studentIds) || studentIds.length === 0) {
+      logger.warn(
+        { eventId, eventTag, examId },
+        "Exam realtime event missing studentIds, skipping notification (core-edu should include studentIds in payload)",
+      );
+      return;
+    }
+
+    const businessEventId = String(
+      rawPayload.eventId ?? rawPayload.event_id ?? eventId,
+    );
+    const groupId =
+      businessEventId.length > 32
+        ? businessEventId.slice(0, 32)
+        : businessEventId;
+
+    for (const userId of studentIds) {
+      await this.notificationsService.send({
+        userId: String(userId),
+        type,
+        title,
+        content,
+        channel: "in_app",
+        groupId,
+        eventId: `${businessEventId}:${userId}`,
+        relatedEntityType: "exam",
+        relatedEntityId: examId,
+        metadata: { source: "core-edu", event: eventTag },
+      });
+    }
+  }
+
  // ============================================================
  // 事件处理器（每个创建对应通知）
  // ============================================================
@@ -279,6 +448,14 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
    if (!Array.isArray(studentIds)) return;
    const examTitle = String(payload.examTitle ?? payload.exam_title ?? "考试");
    const className = String(payload.className ?? payload.class_name ?? "");
+    // groupId 优先用 payload 业务 eventId；降级路径的 Kafka eventId 可能超长，truncate 到 32
+    const businessEventId = String(
+      payload.eventId ?? payload.event_id ?? eventId,
+    );
+    const groupId =
+      businessEventId.length > 32
+        ? businessEventId.slice(0, 32)
+        : businessEventId;

    for (const userId of studentIds) {
      await this.notificationsService.send({
@@ -287,8 +464,8 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
        title: "新考试通知",
        content: `「${className}」班级发布了新考试：${examTitle}`,
        channel: "in_app",
-        groupId: eventId,
-        eventId: `${eventId}:${userId}`,
+        groupId,
+        eventId: `${businessEventId}:${userId}`,
        relatedEntityType: "exam",
        relatedEntityId: String(payload.examId ?? payload.exam_id ?? ""),
        metadata: { source: "core-edu", event: "exam.published" },
@@ -296,6 +473,44 @@ export class KafkaConsumerService implements OnModuleInit, OnModuleDestroy {
    }
  }

+  private async handleHomeworkAssigned(
+    eventId: string,
+    payload: Record<string, unknown>,
+  ): Promise<void> {
+    const studentIds = payload.studentIds ?? payload.student_ids;
+    if (!Array.isArray(studentIds)) return;
+    const homeworkTitle = String(
+      payload.homeworkTitle ?? payload.homework_title ?? "作业",
+    );
+    const className = String(payload.className ?? payload.class_name ?? "");
+    const subject = String(payload.subject ?? "");
+    // groupId 优先用 payload 业务 eventId；降级路径的 Kafka eventId 可能超长，truncate 到 32
+    const businessEventId = String(
+      payload.eventId ?? payload.event_id ?? eventId,
+    );
+    const groupId =
+      businessEventId.length > 32
+        ? businessEventId.slice(0, 32)
+        : businessEventId;
+
+    for (const userId of studentIds) {
+      await this.notificationsService.send({
+        userId: String(userId),
+        type: "homework",
+        title: "新作业通知",
+        content: `${className ? `「${className}」` : ""}${subject ? `${subject}：` : ""}布置了新作业：${homeworkTitle}`,
+        channel: "in_app",
+        groupId,
+        eventId: `${businessEventId}:${userId}`,
+        relatedEntityType: "homework",
+        relatedEntityId: String(
+          payload.homeworkId ?? payload.homework_id ?? "",
+        ),
+        metadata: { source: "core-edu", event: "homework.assigned" },
+      });
+    }
+  }
+
  private async handleAssignmentSubmitted(
    eventId: string,
    payload: Record<string, unknown>,
--- a/services/msg/src/shared/kafka/topic-map.ts
+++ b/services/msg/src/shared/kafka/topic-map.ts
@@ -3,24 +3,24 @@
 *
 * 仲裁依据：
 * - M5：消费 topic 用 `edu.teaching.*` / `edu.identity.*` / `edu.insight.*`
- * - 发布 topic 用 `edu.notification.*`（02-architecture-design.md §5.2）
+ * - ARB-013：发布 topic 用 `edu.notify.notification.*`（统一命名）
 *
 * PRODUCER_TOPIC_MAP：eventType → topic，OutboxPublisher 按此路由发布。
 * CONSUMER_TOPICS：msg 消费的所有 topic 列表，KafkaConsumer 订阅。
 */

-/** 生产者：eventType → Kafka topic 路由 */
+/** 生产者：eventType → Kafka topic 路由（ARB-013 命名） */
 export const PRODUCER_TOPIC_MAP: Record<string, string> = {
-  "notification.sent": "edu.notification.sent",
-  "notification.read": "edu.notification.read",
-  "notification.recalled": "edu.notification.recalled",
-  "notification.failed": "edu.notification.failed",
+  "notification.sent": "edu.notify.notification.sent",
+  "notification.read": "edu.notify.notification.read",
+  "notification.recalled": "edu.notify.notification.recalled",
+  "notification.failed": "edu.notify.notification.failed",
 };

 /** 兜底 topic：未在 TOPIC_MAP 命中的 eventType 走此 topic */
-export const FALLBACK_TOPIC = "edu.notification.events";
+export const FALLBACK_TOPIC = "edu.notify.notification.events";

-/** 消费者：订阅的 topic 列表（iam 6 + core-edu 5 + data-ana 1 = 12 类事件） */
+/** 消费者：订阅的 topic 列表（iam 6 + core-edu 9 + data-ana 1 = 16 类事件） */
 export const CONSUMER_TOPICS: readonly string[] = [
  // iam（identity）
  "edu.identity.user.created",
@@ -29,12 +29,16 @@ export const CONSUMER_TOPICS: readonly string[] = [
  "edu.identity.user.role_changed",
  "edu.identity.role.created",
  "edu.identity.role.updated",
-  // core-edu（teaching）
+  // core-edu（teaching）— 6 基础 + 3 考试实时事件
  "edu.teaching.exam.published",
+  "edu.teaching.homework.assigned",
  "edu.teaching.assignment.submitted",
  "edu.teaching.assignment.graded",
  "edu.teaching.grade.recorded",
  "edu.teaching.attendance.recorded",
+  "edu.teaching.exam.extended",
+  "edu.teaching.exam.force_submitted",
+  "edu.teaching.exam.question_reordered",
  // data-ana（insight）
  "edu.insight.mastery.updated",
 ] as const;
@@ -46,6 +50,7 @@ export const CONSUMER_EVENT_TYPE_MAP: Record<string, string> = {
  "edu.identity.role.created": "system",
  "edu.identity.role.updated": "system",
  "edu.teaching.exam.published": "exam",
+  "edu.teaching.homework.assigned": "homework",
  "edu.teaching.assignment.submitted": "homework",
  "edu.teaching.assignment.graded": "grade",
  "edu.teaching.grade.recorded": "grade",
--- a/services/msg/src/shared/observability/logger.ts
+++ b/services/msg/src/shared/observability/logger.ts
@@ -1,17 +1,17 @@
-import pino from 'pino';
-import { env } from '../../config/env.js';
+import { pino } from "pino";
+import { env } from "../../config/env.js";

 export const logger = pino({
  level: env.LOG_LEVEL,
  // 修复：pino 默认字段选项为 `base`，而非 `defaultFields`
  base: {
-    service: 'msg',
-    version: '0.1.0',
+    service: "msg",
+    version: "0.1.0",
  },
  transport:
-    env.NODE_ENV === 'development'
+    env.NODE_ENV === "development"
      ? {
-          target: 'pino-pretty',
+          target: "pino-pretty",
          options: { colorize: true },
        }
      : undefined,
--- a/services/msg/src/shared/outbox/outbox.publisher.ts
+++ b/services/msg/src/shared/outbox/outbox.publisher.ts
@@ -13,10 +13,10 @@ import type { OutboxEvent } from "./outbox.schema.js";
 * OutboxPublisher —— 轮询 pending 记录并投递到 Kafka（多 topic 路由）。
 *
 * 参照 core-edu OutboxPublisher 模式，支持 TOPIC_MAP 多 topic 路由：
- * - notification.sent → edu.notification.sent
- * - notification.read → edu.notification.read
- * - notification.recalled → edu.notification.recalled
- * - notification.failed → edu.notification.failed
+ * - 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 去重）。
 */
--- a/services/msg/src/shared/outbox/outbox.schema.ts
+++ b/services/msg/src/shared/outbox/outbox.schema.ts
@@ -25,7 +25,7 @@ export type OutboxStatus = "pending" | "published" | "failed";
 * msg_outbox_events 表：事务性 Outbox。
 *
 * 业务事务内写入，OutboxPublisher 轮询 pending 记录投递到 Kafka。
- * eventType 经 TOPIC_MAP 路由到不同 topic（edu.notification.sent/read/recalled/failed）。
+ * eventType 经 TOPIC_MAP 路由到不同 topic（edu.notify.notification.sent/read/recalled/failed）。
 */
 export const outboxEvents = mysqlTable("msg_outbox_events", {
  eventId: varchar("event_id", { length: 64 }).notNull().primaryKey(),