feat(core-edu): v2 P3.14 考试实时事件 + pino 修复
新增 3 RPC：ExtendExam/ForceSubmitExam/ReorderExamQuestions
新增 3 Kafka 事件：exam.extended/exam.force_submitted/exam.question_reordered
exams.service.ts 新增 3 方法 + Outbox 事务内写入 + TOPIC_MAP 映射
grpc.server.ts 注册 3 handler + grpc-smoke 测试
logger.ts pino 导入修复（import pino → import { pino }）
27/27 smoke test 通过
This commit is contained in:
SpecialX
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
|
||||
import { eq, and } from "drizzle-orm";
|
||||
import { Injectable } from "@nestjs/common";
|
||||
import { db } from "../config/database.js";
|
||||
import { exams } from "./exams.schema.js";
|
||||
import { exams, examQuestions, examSubmissions } from "./exams.schema.js";
|
||||
import { examDrafts, examViolations } from "./exam-extensions.schema.js";
|
||||
import { examsRepository } from "./exams.repository.js";
|
||||
import { outboxRepository } from "../shared/outbox/outbox.repository.js";
|
||||
@@ -478,6 +478,215 @@ export class ExamsService {
|
||||
return { violationId };
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------------------
|
||||
// P3.14 新增：考试实时事件（延长考试 / 强制收卷 / 调整题目顺序）
|
||||
// 事件经 Outbox → Kafka → msg → push-gateway → student-portal WebSocket
|
||||
// --------------------------------------------------------------------------
|
||||
|
||||
async extendExam(
|
||||
examId: string,
|
||||
extensionSeconds: number,
|
||||
extendedBy: string,
|
||||
): Promise<{ success: boolean; newDuration: number }> {
|
||||
if (!examId || extensionSeconds <= 0) {
|
||||
throw new ValidationError(
|
||||
"examId and positive extensionSeconds are required",
|
||||
);
|
||||
}
|
||||
const exam = await examsRepository.findById(examId);
|
||||
if (!exam) {
|
||||
throw new NotFoundError(`Exam ${examId} not found`);
|
||||
}
|
||||
// 只有 published 或 in_progress 状态的考试可以延长
|
||||
if (exam.status !== "published" && exam.status !== "in_progress") {
|
||||
throw new ConflictError(
|
||||
`Exam ${examId} status ${exam.status} cannot be extended`,
|
||||
);
|
||||
}
|
||||
|
||||
const newDuration = exam.duration + extensionSeconds;
|
||||
await db.transaction(async (tx) => {
|
||||
await tx
|
||||
.update(exams)
|
||||
.set({ duration: newDuration, updatedAt: new Date() })
|
||||
.where(eq(exams.id, examId));
|
||||
const event = buildEvent({
|
||||
aggregateId: examId,
|
||||
eventType: "exam.extended",
|
||||
payload: {
|
||||
examId,
|
||||
classId: exam.classId,
|
||||
subjectId: exam.subjectId,
|
||||
extensionSeconds,
|
||||
newDuration,
|
||||
},
|
||||
userId: extendedBy,
|
||||
});
|
||||
await outboxRepository.create(
|
||||
{
|
||||
id: randomUUID(),
|
||||
eventId: event.event_id,
|
||||
aggregateId: examId,
|
||||
aggregateType: "exam",
|
||||
eventType: "exam.extended",
|
||||
occurredAt: new Date(event.occurred_at),
|
||||
payload: serializeEvent(event),
|
||||
status: "pending",
|
||||
},
|
||||
tx,
|
||||
);
|
||||
});
|
||||
return { success: true, newDuration };
|
||||
}
|
||||
|
||||
async forceSubmitExam(
|
||||
examId: string,
|
||||
studentIds: string[],
|
||||
forcedBy: string,
|
||||
): Promise<{ affectedCount: number }> {
|
||||
if (!examId) {
|
||||
throw new ValidationError("examId is required");
|
||||
}
|
||||
const exam = await examsRepository.findById(examId);
|
||||
if (!exam) {
|
||||
throw new NotFoundError(`Exam ${examId} not found`);
|
||||
}
|
||||
if (exam.status !== "in_progress") {
|
||||
throw new ConflictError(
|
||||
`Exam ${examId} status ${exam.status} cannot be force submitted`,
|
||||
);
|
||||
}
|
||||
|
||||
// 查询需要强制收卷的提交记录
|
||||
let targetStudentIds = studentIds;
|
||||
if (targetStudentIds.length === 0) {
|
||||
// 空列表 = 全部未提交学生
|
||||
const pending = await db
|
||||
.select({
|
||||
id: examSubmissions.id,
|
||||
studentId: examSubmissions.studentId,
|
||||
})
|
||||
.from(examSubmissions)
|
||||
.where(
|
||||
and(
|
||||
eq(examSubmissions.examId, examId),
|
||||
eq(examSubmissions.status, "not_submitted"),
|
||||
),
|
||||
);
|
||||
targetStudentIds = pending.map((p) => p.studentId);
|
||||
}
|
||||
|
||||
let affectedCount = 0;
|
||||
if (targetStudentIds.length > 0) {
|
||||
await db.transaction(async (tx) => {
|
||||
for (const studentId of targetStudentIds) {
|
||||
const result = await tx
|
||||
.update(examSubmissions)
|
||||
.set({
|
||||
status: "submitted",
|
||||
submittedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(examSubmissions.examId, examId),
|
||||
eq(examSubmissions.studentId, studentId),
|
||||
eq(examSubmissions.status, "not_submitted"),
|
||||
),
|
||||
);
|
||||
affectedCount += result[0]?.affectedRows ?? 0;
|
||||
}
|
||||
const event = buildEvent({
|
||||
aggregateId: examId,
|
||||
eventType: "exam.force_submitted",
|
||||
payload: {
|
||||
examId,
|
||||
classId: exam.classId,
|
||||
subjectId: exam.subjectId,
|
||||
studentIds: targetStudentIds,
|
||||
affectedCount,
|
||||
},
|
||||
userId: forcedBy,
|
||||
});
|
||||
await outboxRepository.create(
|
||||
{
|
||||
id: randomUUID(),
|
||||
eventId: event.event_id,
|
||||
aggregateId: examId,
|
||||
aggregateType: "exam",
|
||||
eventType: "exam.force_submitted",
|
||||
occurredAt: new Date(event.occurred_at),
|
||||
payload: serializeEvent(event),
|
||||
status: "pending",
|
||||
},
|
||||
tx,
|
||||
);
|
||||
});
|
||||
}
|
||||
return { affectedCount };
|
||||
}
|
||||
|
||||
async reorderExamQuestions(
|
||||
examId: string,
|
||||
orders: Array<{ questionId: string; order: number }>,
|
||||
reorderedBy: string,
|
||||
): Promise<{ success: boolean }> {
|
||||
if (!examId || orders.length === 0) {
|
||||
throw new ValidationError("examId and non-empty orders are required");
|
||||
}
|
||||
const exam = await examsRepository.findById(examId);
|
||||
if (!exam) {
|
||||
throw new NotFoundError(`Exam ${examId} not found`);
|
||||
}
|
||||
// 只有 draft 或 published 状态可以调整题目顺序
|
||||
if (exam.status !== "draft" && exam.status !== "published") {
|
||||
throw new ConflictError(
|
||||
`Exam ${examId} status ${exam.status} cannot reorder questions`,
|
||||
);
|
||||
}
|
||||
|
||||
await db.transaction(async (tx) => {
|
||||
for (const item of orders) {
|
||||
await tx
|
||||
.update(examQuestions)
|
||||
.set({ order: item.order })
|
||||
.where(
|
||||
and(
|
||||
eq(examQuestions.examId, examId),
|
||||
eq(examQuestions.questionId, item.questionId),
|
||||
),
|
||||
);
|
||||
}
|
||||
const event = buildEvent({
|
||||
aggregateId: examId,
|
||||
eventType: "exam.question_reordered",
|
||||
payload: {
|
||||
examId,
|
||||
classId: exam.classId,
|
||||
subjectId: exam.subjectId,
|
||||
orders: orders.map((o) => ({
|
||||
questionId: o.questionId,
|
||||
order: o.order,
|
||||
})),
|
||||
},
|
||||
userId: reorderedBy,
|
||||
});
|
||||
await outboxRepository.create(
|
||||
{
|
||||
id: randomUUID(),
|
||||
eventId: event.event_id,
|
||||
aggregateId: examId,
|
||||
aggregateType: "exam",
|
||||
eventType: "exam.question_reordered",
|
||||
occurredAt: new Date(event.occurred_at),
|
||||
payload: serializeEvent(event),
|
||||
status: "pending",
|
||||
},
|
||||
tx,
|
||||
);
|
||||
});
|
||||
return { success: true };
|
||||
}
|
||||
|
||||
private assertTransition(from: ExamStatus, action: ExamAction): void {
|
||||
if (!canTransition(from, action)) {
|
||||
throw new ApplicationError(
|
||||
|
||||
@@ -426,6 +426,38 @@ function buildExamHandlers(
|
||||
);
|
||||
return { violation_id: result.violationId };
|
||||
}),
|
||||
|
||||
ExtendExam: wrapHandler(async (req) => {
|
||||
const result = await service.extendExam(
|
||||
reqStr(req, "exam_id"),
|
||||
reqNum(req, "extension_seconds"),
|
||||
reqStr(req, "extended_by"),
|
||||
);
|
||||
return { success: result.success, new_duration: result.newDuration };
|
||||
}),
|
||||
|
||||
ForceSubmitExam: wrapHandler(async (req) => {
|
||||
const studentIds = reqStrArr(req, "student_ids");
|
||||
const result = await service.forceSubmitExam(
|
||||
reqStr(req, "exam_id"),
|
||||
studentIds,
|
||||
reqStr(req, "forced_by"),
|
||||
);
|
||||
return { affected_count: result.affectedCount };
|
||||
}),
|
||||
|
||||
ReorderExamQuestions: wrapHandler(async (req) => {
|
||||
const orders = reqObjArr(req, "orders").map((o) => ({
|
||||
questionId: reqStr(o, "question_id"),
|
||||
order: reqNum(o, "order"),
|
||||
}));
|
||||
const result = await service.reorderExamQuestions(
|
||||
reqStr(req, "exam_id"),
|
||||
orders,
|
||||
reqStr(req, "reordered_by"),
|
||||
);
|
||||
return { success: result.success };
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
import pino from 'pino';
|
||||
import { env } from '../../config/env.js';
|
||||
import { pino } from "pino";
|
||||
import { env } from "../../config/env.js";
|
||||
|
||||
export const logger = pino({
|
||||
name: 'core-edu',
|
||||
name: "core-edu",
|
||||
level: env.LOG_LEVEL,
|
||||
base: { service: 'core-edu' },
|
||||
...(env.NODE_ENV === 'development'
|
||||
base: { service: "core-edu" },
|
||||
...(env.NODE_ENV === "development"
|
||||
? {
|
||||
transport: {
|
||||
target: 'pino-pretty',
|
||||
options: { colorize: true, translateTime: 'SYS:standard' },
|
||||
target: "pino-pretty",
|
||||
options: { colorize: true, translateTime: "SYS:standard" },
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
|
||||
@@ -17,6 +17,10 @@ const TOPIC_MAP: Record<string, string> = {
|
||||
"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",
|
||||
|
||||
Reference in New Issue
Block a user