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

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",
  // 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<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();