feat(core-edu): 完整实现 core-edu 教学核心服务

包含 classes/exams/homework/grades/attendance/scheduling 域、outbox、iam-consumer、redis 配置等完整实现

services/core-edu/src/shared/outbox/event-builder.ts

@@ -0,0 +1,61 @@
+import { randomUUID } from "node:crypto";
+
+/**
+ * Outbox 事件构建器
+ *
+ * 所有事件 payload 必须包含（events.proto 仲裁 §2.1）：
+ *   - event_id: UUID（幂等消费）
+ *   - occurred_at: 业务时间戳（ms epoch）
+ *   - schema_version: 默认 "v1"
+ *   - metadata: { traceId, userId }
+ */
+
+export interface EventMetadata {
+  schema_version: string;
+  trace_id: string;
+  user_id: string;
+}
+
+export interface OutboxEvent<T = Record<string, unknown>> {
+  event_id: string;
+  aggregate_id: string;
+  event_type: string;
+  occurred_at: number;
+  payload: T;
+  metadata: EventMetadata;
+}
+
+export interface BuildEventInput {
+  aggregateId: string;
+  eventType: string;
+  payload: Record<string, unknown>;
+  userId?: string;
+  traceId?: string;
+  schemaVersion?: string;
+}
+
+/**
+ * 构建标准 Outbox 事件（含 event_id / occurred_at / schema_version / metadata）。
+ * 返回的对象可直接 JSON.stringify 作为 outbox.payload 存储。
+ */
+export function buildEvent(input: BuildEventInput): OutboxEvent {
+  return {
+    event_id: randomUUID(),
+    aggregate_id: input.aggregateId,
+    event_type: input.eventType,
+    occurred_at: Date.now(),
+    payload: input.payload,
+    metadata: {
+      schema_version: input.schemaVersion ?? "v1",
+      trace_id: input.traceId ?? "unknown",
+      user_id: input.userId ?? "system",
+    },
+  };
+}
+
+/**
+ * 将 OutboxEvent 序列化为 JSON 字符串（用于 outbox.payload 列）。
+ */
+export function serializeEvent(event: OutboxEvent): string {
+  return JSON.stringify(event);
+}

services/core-edu/src/shared/outbox/outbox.publisher.ts

@@ -1,30 +1,46 @@
-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';
+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.created': 'edu.exam.events',
-  'exam.updated': 'edu.exam.events',
-  'exam.deleted': 'edu.exam.events',
-  'homework.assigned': 'edu.homework.events',
-  'homework.submitted': 'edu.homework.events',
-  'homework.graded': 'edu.homework.events',
-  'grade.recorded': 'edu.grade.events',
-  'grade.updated': 'edu.grade.events',
-  'class.transferred': 'edu.class.events',
+  // 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",
+  // 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');
+    logger.info("OutboxPublisher started");
     this.intervalId = setInterval(() => {
       void this.poll();
     }, POLL_INTERVAL_MS);
@@ -35,7 +51,7 @@ export class OutboxPublisher {
       clearInterval(this.intervalId);
       this.intervalId = null;
     }
-    logger.info('OutboxPublisher stopped');
+    logger.info("OutboxPublisher stopped");
   }
 
   private async poll(): Promise<void> {
@@ -47,14 +63,14 @@ export class OutboxPublisher {
         await this.publish(message);
       }
     } catch (error) {
-      logger.error({ error }, 'Outbox poll failed');
+      logger.error({ error }, "Outbox poll failed");
     } finally {
       this.isPolling = false;
     }
   }
 
   private async publish(message: OutboxMessage): Promise<void> {
-    const topic = TOPIC_MAP[message.eventType] ?? 'edu.fallback.events';
+    const topic = TOPIC_MAP[message.eventType] ?? "edu.teaching.fallback";
     try {
       await producer.send({
         topic,
@@ -65,21 +81,32 @@ export class OutboxPublisher {
             headers: {
               eventType: message.eventType,
               aggregateType: message.aggregateType,
+              eventId: message.eventId,
             },
           },
         ],
       });
       await outboxRepository.markProcessed(message.id);
       logger.info(
-        { id: message.id, eventType: message.eventType, topic },
-        'Outbox message published',
+        {
+          id: message.id,
+          eventId: message.eventId,
+          eventType: message.eventType,
+          topic,
+        },
+        "Outbox message published",
       );
     } catch (error) {
-      logger.error({ error, id: message.id }, 'Outbox publish failed');
-      if (message.retryCount + 1 >= MAX_RETRY) {
+      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 {
-        await outboxRepository.incrementRetry(message.id);
+        const backoff = RETRY_BACKOFF_BASE_MS * Math.pow(2, nextRetry);
+        await outboxRepository.incrementRetry(message.id, backoff);
       }
     }
   }

services/core-edu/src/shared/outbox/outbox.repository.ts

@@ -1,6 +1,10 @@
-import { eq, sql } from 'drizzle-orm';
-import { db } from '../../config/database.js';
-import { outbox, type OutboxMessage, type NewOutboxMessage } from './outbox.schema.js';
+import { eq, sql, and, lte, isNull, or } from "drizzle-orm";
+import { db } from "../../config/database.js";
+import {
+  outbox,
+  type OutboxMessage,
+  type NewOutboxMessage,
+} from "./outbox.schema.js";
 
 type DbClient = typeof db;
 
@@ -9,33 +13,45 @@ export class OutboxRepository {
     await tx.insert(outbox).values(message);
   }
 
+  /**
+   * 查找待发布消息：
+   * - status = 'pending'
+   * - next_retry_at IS NULL 或 next_retry_at <= NOW()
+   * 按 created_at 升序，取 limit 条
+   */
   async findPending(limit: number = 100): Promise<OutboxMessage[]> {
     return db
       .select()
       .from(outbox)
-      .where(eq(outbox.status, 'pending'))
+      .where(
+        and(
+          eq(outbox.status, "pending"),
+          or(lte(outbox.nextRetryAt, new Date()), isNull(outbox.nextRetryAt)),
+        ),
+      )
       .limit(limit);
   }
 
   async markProcessed(id: string): Promise<void> {
     await db
       .update(outbox)
-      .set({ status: 'processed', processedAt: new Date() })
+      .set({ status: "processed", processedAt: new Date() })
       .where(eq(outbox.id, id));
   }
 
-  async incrementRetry(id: string): Promise<void> {
+  async incrementRetry(id: string, backoffMs: number): Promise<void> {
+    const nextRetryAt = new Date(Date.now() + backoffMs);
     await db
       .update(outbox)
-      .set({ retryCount: sql`${outbox.retryCount} + 1` })
+      .set({
+        retryCount: sql`${outbox.retryCount} + 1`,
+        nextRetryAt,
+      })
       .where(eq(outbox.id, id));
   }
 
   async markFailed(id: string): Promise<void> {
-    await db
-      .update(outbox)
-      .set({ status: 'failed' })
-      .where(eq(outbox.id, id));
+    await db.update(outbox).set({ status: "failed" }).where(eq(outbox.id, id));
   }
 }
 

services/core-edu/src/shared/outbox/outbox.schema.ts

@@ -1,16 +1,44 @@
-import { mysqlTable, varchar, text, timestamp, char, bigint } from 'drizzle-orm/mysql-core';
+import {
+  mysqlTable,
+  varchar,
+  text,
+  timestamp,
+  char,
+  bigint,
+  datetime,
+  index,
+  uniqueIndex,
+} from "drizzle-orm/mysql-core";
 
-export const outbox = mysqlTable('core_edu_outbox', {
-  id: char('id', { length: 36 }).notNull().primaryKey(),
-  aggregateId: char('aggregate_id', { length: 36 }).notNull(),
-  aggregateType: varchar('aggregate_type', { length: 50 }).notNull(),
-  eventType: varchar('event_type', { length: 100 }).notNull(),
-  payload: text('payload').notNull(),
-  status: varchar('status', { length: 20 }).notNull().default('pending'),
-  retryCount: bigint('retry_count', { mode: 'number' }).notNull().default(0),
-  createdAt: timestamp('created_at').notNull().defaultNow(),
-  processedAt: timestamp('processed_at'),
-});
+// 事务性发件箱（Outbox 模式）
+export const outbox = mysqlTable(
+  "core_edu_outbox",
+  {
+    id: char("id", { length: 36 }).notNull().primaryKey(),
+    eventId: char("event_id", { length: 36 }).notNull(),
+    aggregateId: char("aggregate_id", { length: 36 }).notNull(),
+    aggregateType: varchar("aggregate_type", { length: 50 }).notNull(),
+    eventType: varchar("event_type", { length: 100 }).notNull(),
+    occurredAt: datetime("occurred_at").notNull(),
+    payload: text("payload").notNull(),
+    status: varchar("status", { length: 20 }).notNull().default("pending"),
+    retryCount: bigint("retry_count", { mode: "number" }).notNull().default(0),
+    nextRetryAt: datetime("next_retry_at"),
+    createdAt: timestamp("created_at").notNull().defaultNow(),
+    processedAt: timestamp("processed_at"),
+  },
+  (table) => ({
+    uniqEventId: uniqueIndex("uniq_event_id").on(table.eventId),
+    idxOutboxStatusRetry: index("idx_outbox_status_retry").on(
+      table.status,
+      table.nextRetryAt,
+    ),
+    idxOutboxAggregate: index("idx_outbox_aggregate").on(
+      table.aggregateType,
+      table.aggregateId,
+    ),
+  }),
+);
 
 export type OutboxMessage = typeof outbox.$inferSelect;
 export type NewOutboxMessage = typeof outbox.$inferInsert;