import { Injectable, OnModuleInit, Logger } from '@nestjs/common';
import { CDCConsumerService, CDCEvent } from '../../infrastructure/kafka/cdc-consumer.service';
import { UserSyncedHandler } from './user-synced.handler';
import { ReferralSyncedHandler } from './referral-synced.handler';
import { AdoptionSyncedHandler } from './adoption-synced.handler';

/**
 * CDC 事件分发器
 * 负责将 Debezium CDC 事件路由到对应的处理器
 */
@Injectable()
export class CDCEventDispatcher implements OnModuleInit {
  private readonly logger = new Logger(CDCEventDispatcher.name);

  constructor(
    private readonly cdcConsumer: CDCConsumerService,
    private readonly userHandler: UserSyncedHandler,
    private readonly referralHandler: ReferralSyncedHandler,
    private readonly adoptionHandler: AdoptionSyncedHandler,
  ) {}

  async onModuleInit() {
    // 注册各表的事件处理器
    this.cdcConsumer.registerHandler('users', this.handleUserEvent.bind(this));
    this.cdcConsumer.registerHandler('referrals', this.handleReferralEvent.bind(this));
    this.cdcConsumer.registerHandler('adoptions', this.handleAdoptionEvent.bind(this));

    // 启动 CDC 消费者
    try {
      await this.cdcConsumer.start();
      this.logger.log('CDC event dispatcher started');
    } catch (error) {
      this.logger.error('Failed to start CDC event dispatcher', error);
      // 不抛出错误，允许服务在没有 Kafka 的情况下启动（用于本地开发）
    }
  }

  private async handleUserEvent(event: CDCEvent): Promise<void> {
    await this.userHandler.handle(event);
  }

  private async handleReferralEvent(event: CDCEvent): Promise<void> {
    await this.referralHandler.handle(event);
  }

  private async handleAdoptionEvent(event: CDCEvent): Promise<void> {
    await this.adoptionHandler.handle(event);
  }
}
