58c0ba1bd9
包含 classes/exams/homework/grades/attendance/scheduling 域、outbox、iam-consumer、redis 配置等完整实现
116 lines
3.4 KiB
TypeScript
116 lines
3.4 KiB
TypeScript
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",
|
||
// 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();
|