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';

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',
};

const POLL_INTERVAL_MS = 5000;
const BATCH_SIZE = 100;
const MAX_RETRY = 5;

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.fallback.events';
    try {
      await producer.send({
        topic,
        messages: [
          {
            key: message.aggregateId,
            value: message.payload,
            headers: {
              eventType: message.eventType,
              aggregateType: message.aggregateType,
            },
          },
        ],
      });
      await outboxRepository.markProcessed(message.id);
      logger.info(
        { id: message.id, eventType: message.eventType, topic },
        'Outbox message published',
      );
    } catch (error) {
      logger.error({ error, id: message.id }, 'Outbox publish failed');
      if (message.retryCount + 1 >= MAX_RETRY) {
        await outboxRepository.markFailed(message.id);
      } else {
        await outboxRepository.incrementRetry(message.id);
      }
    }
  }
}

export const outboxPublisher = new OutboxPublisher();