import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

/**
 * CDC 事件结构 (Debezium 格式)
 */
export interface CdcEvent {
  schema: any;
  payload: {
    before: any | null;
    after: any | null;
    source: {
      version: string;
      connector: string;
      name: string;
      ts_ms: number;
      snapshot: string;
      db: string;
      sequence: string;
      schema: string;
      table: string;
      txId: number;
      lsn: number;
      xmin: number | null;
    };
    op: 'c' | 'u' | 'd' | 'r'; // create, update, delete, read (snapshot)
    ts_ms: number;
    transaction: any;
  };
  sequenceNum: bigint;
}

/**
 * 2.0 服务间事件结构 (Outbox 格式)
 */
export interface ServiceEvent {
  id: string;
  aggregateType: string;
  aggregateId: string;
  eventType: string;
  payload: any;
  createdAt: string;
  sequenceNum: bigint;
}

export type CdcHandler = (event: CdcEvent) => Promise<void>;
export type ServiceEventHandler = (event: ServiceEvent) => Promise<void>;

@Injectable()
export class CdcConsumerService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(CdcConsumerService.name);
  private kafka: Kafka;
  private consumer: Consumer;
  private cdcHandlers: Map<string, CdcHandler> = new Map();
  private serviceHandlers: Map<string, ServiceEventHandler> = new Map();
  private isRunning = false;
  private topics: string[] = [];

  constructor(private readonly configService: ConfigService) {
    const brokers = this.configService
      .get<string>('KAFKA_BROKERS', 'localhost:9092')
      .split(',');

    this.kafka = new Kafka({
      clientId: 'mining-admin-service-cdc',
      brokers,
    });

    this.consumer = this.kafka.consumer({
      groupId: this.configService.get<string>(
        'CDC_CONSUMER_GROUP',
        'mining-admin-service-cdc-group',
      ),
    });
  }

  async onModuleInit() {
    // 启动延迟到 CdcSyncService 注册完处理器后
  }

  async onModuleDestroy() {
    await this.stop();
  }

  /**
   * 注册 CDC 事件处理器 (1.0 → 2.0 同步)
   */
  registerCdcHandler(tableName: string, handler: CdcHandler): void {
    this.cdcHandlers.set(tableName, handler);
    this.logger.log(`Registered CDC handler for table: ${tableName}`);
  }

  /**
   * 注册服务事件处理器 (2.0 服务间同步)
   */
  registerServiceHandler(eventType: string, handler: ServiceEventHandler): void {
    this.serviceHandlers.set(eventType, handler);
    this.logger.log(`Registered service event handler for: ${eventType}`);
  }

  /**
   * 添加要订阅的 topic
   */
  addTopic(topic: string): void {
    if (!this.topics.includes(topic)) {
      this.topics.push(topic);
    }
  }

  /**
   * 启动消费者
   */
  async start(): Promise<void> {
    if (this.isRunning) {
      this.logger.warn('CDC consumer is already running');
      return;
    }

    if (this.topics.length === 0) {
      this.logger.warn('No topics to subscribe, skipping CDC consumer start');
      return;
    }

    try {
      await this.consumer.connect();
      this.logger.log('CDC consumer connected');

      await this.consumer.subscribe({
        topics: this.topics,
        fromBeginning: false,
      });
      this.logger.log(`Subscribed to topics: ${this.topics.join(', ')}`);

      await this.consumer.run({
        eachMessage: async (payload: EachMessagePayload) => {
          await this.handleMessage(payload);
        },
      });

      this.isRunning = true;
      this.logger.log('CDC consumer started');
    } catch (error) {
      this.logger.error('Failed to start CDC consumer', error);
      // 不抛出错误，允许服务继续运行（CDC 可能暂时不可用）
    }
  }

  /**
   * 停止消费者
   */
  async stop(): Promise<void> {
    if (!this.isRunning) {
      return;
    }

    try {
      await this.consumer.disconnect();
      this.isRunning = false;
      this.logger.log('CDC consumer stopped');
    } catch (error) {
      this.logger.error('Failed to stop CDC consumer', error);
    }
  }

  private async handleMessage(payload: EachMessagePayload): Promise<void> {
    const { topic, partition, message } = payload;

    try {
      if (!message.value) {
        return;
      }

      const eventData = JSON.parse(message.value.toString());
      const sequenceNum = BigInt(message.offset);

      // 判断事件类型：Debezium CDC 或 服务 Outbox 事件
      if (this.isDebeziumEvent(eventData)) {
        await this.handleCdcEvent(topic, eventData, sequenceNum);
      } else if (this.isServiceEvent(eventData)) {
        await this.handleServiceEvent(topic, eventData, sequenceNum);
      } else {
        this.logger.warn(`Unknown event format from topic: ${topic}`);
      }
    } catch (error) {
      this.logger.error(
        `Error processing message from topic ${topic}, partition ${partition}`,
        error,
      );
    }
  }

  private isDebeziumEvent(data: any): boolean {
    return data.payload && data.payload.source && data.payload.op;
  }

  private isServiceEvent(data: any): boolean {
    return data.eventType && data.aggregateType;
  }

  private async handleCdcEvent(
    topic: string,
    eventData: any,
    sequenceNum: bigint,
  ): Promise<void> {
    const event: CdcEvent = {
      ...eventData,
      sequenceNum,
    };

    // 从 topic 名称提取表名 (格式: dbserver1.schema.tablename)
    const parts = topic.split('.');
    const tableName = parts[parts.length - 1];

    const handler = this.cdcHandlers.get(tableName);
    if (handler) {
      await handler(event);
      this.logger.debug(
        `Processed CDC event for table ${tableName}, op: ${event.payload.op}`,
      );
    }
  }

  private async handleServiceEvent(
    topic: string,
    eventData: any,
    sequenceNum: bigint,
  ): Promise<void> {
    const event: ServiceEvent = {
      ...eventData,
      sequenceNum,
    };

    const handler = this.serviceHandlers.get(event.eventType);
    if (handler) {
      await handler(event);
      this.logger.debug(`Processed service event: ${event.eventType}`);
    } else {
      // 尝试通配符处理器
      const aggregateHandler = this.serviceHandlers.get(
        `${event.aggregateType}.*`,
      );
      if (aggregateHandler) {
        await aggregateHandler(event);
        this.logger.debug(
          `Processed service event via wildcard: ${event.eventType}`,
        );
      }
    }
  }
}
