import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { Cron, CronExpression } from '@nestjs/schedule';
import { ContributionCalculationService } from '../services/contribution-calculation.service';
import { SnapshotService } from '../services/snapshot.service';
import { ContributionRecordRepository } from '../../infrastructure/persistence/repositories/contribution-record.repository';
import { OutboxRepository } from '../../infrastructure/persistence/repositories/outbox.repository';
import { KafkaProducerService } from '../../infrastructure/kafka/kafka-producer.service';
import { RedisService } from '../../infrastructure/redis/redis.service';

/**
 * 算力相关定时任务
 */
@Injectable()
export class ContributionScheduler implements OnModuleInit {
  private readonly logger = new Logger(ContributionScheduler.name);
  private readonly LOCK_KEY = 'contribution:scheduler:lock';

  constructor(
    private readonly calculationService: ContributionCalculationService,
    private readonly snapshotService: SnapshotService,
    private readonly contributionRecordRepository: ContributionRecordRepository,
    private readonly outboxRepository: OutboxRepository,
    private readonly kafkaProducer: KafkaProducerService,
    private readonly redis: RedisService,
  ) {}

  async onModuleInit() {
    this.logger.log('Contribution scheduler initialized');
  }

  /**
   * 每分钟处理未处理的认种记录
   */
  @Cron(CronExpression.EVERY_MINUTE)
  async processUnprocessedAdoptions(): Promise<void> {
    const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:process`, 55);
    if (!lockValue) {
      return; // 其他实例正在处理
    }

    try {
      const processed = await this.calculationService.processUndistributedAdoptions(100);
      if (processed > 0) {
        this.logger.log(`Processed ${processed} unprocessed adoptions`);
      }
    } catch (error) {
      this.logger.error('Failed to process unprocessed adoptions', error);
    } finally {
      await this.redis.releaseLock(`${this.LOCK_KEY}:process`, lockValue);
    }
  }

  /**
   * 每天凌晨1点创建每日快照
   */
  @Cron('0 1 * * *')
  async createDailySnapshot(): Promise<void> {
    const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:snapshot`, 300);
    if (!lockValue) {
      return;
    }

    try {
      // 创建前一天的快照
      const yesterday = new Date();
      yesterday.setDate(yesterday.getDate() - 1);

      await this.snapshotService.createDailySnapshot(yesterday);
      this.logger.log(`Daily snapshot created for ${yesterday.toISOString().split('T')[0]}`);
    } catch (error) {
      this.logger.error('Failed to create daily snapshot', error);
    } finally {
      await this.redis.releaseLock(`${this.LOCK_KEY}:snapshot`, lockValue);
    }
  }

  /**
   * 每天凌晨2点检查过期的算力记录
   */
  @Cron('0 2 * * *')
  async processExpiredRecords(): Promise<void> {
    const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:expire`, 300);
    if (!lockValue) {
      return;
    }

    try {
      const now = new Date();
      const expiredRecords = await this.contributionRecordRepository.findExpiredRecords(now, 1000);

      if (expiredRecords.length > 0) {
        const ids = expiredRecords.map((r) => r.id).filter((id): id is bigint => id !== null);
        await this.contributionRecordRepository.markAsExpired(ids);
        this.logger.log(`Marked ${ids.length} contribution records as expired`);

        // TODO: 需要相应地减少账户的算力值
      }
    } catch (error) {
      this.logger.error('Failed to process expired records', error);
    } finally {
      await this.redis.releaseLock(`${this.LOCK_KEY}:expire`, lockValue);
    }
  }

  /**
   * 每30秒发布 Outbox 中的事件
   */
  @Cron('*/30 * * * * *')
  async publishOutboxEvents(): Promise<void> {
    const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:outbox`, 25);
    if (!lockValue) {
      return;
    }

    try {
      const events = await this.outboxRepository.findUnprocessed(100);

      if (events.length === 0) {
        return;
      }

      for (const event of events) {
        try {
          await this.kafkaProducer.emit(`contribution.${event.eventType}`, {
            key: event.aggregateId,
            value: {
              eventId: event.id,
              aggregateType: event.aggregateType,
              aggregateId: event.aggregateId,
              eventType: event.eventType,
              payload: event.payload,
              createdAt: event.createdAt.toISOString(),
            },
          });
        } catch (error) {
          this.logger.error(`Failed to publish event ${event.id}`, error);
          // 继续处理下一个事件
          continue;
        }
      }

      // 标记为已处理
      const processedIds = events.map((e) => e.id);
      await this.outboxRepository.markAsProcessed(processedIds);

      this.logger.debug(`Published ${processedIds.length} outbox events`);
    } catch (error) {
      this.logger.error('Failed to publish outbox events', error);
    } finally {
      await this.redis.releaseLock(`${this.LOCK_KEY}:outbox`, lockValue);
    }
  }

  /**
   * 每天凌晨3点清理已处理的 Outbox 事件（保留7天）
   */
  @Cron('0 3 * * *')
  async cleanupOutbox(): Promise<void> {
    const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:cleanup`, 300);
    if (!lockValue) {
      return;
    }

    try {
      const sevenDaysAgo = new Date();
      sevenDaysAgo.setDate(sevenDaysAgo.getDate() - 7);

      const deleted = await this.outboxRepository.deleteProcessed(sevenDaysAgo);
      if (deleted > 0) {
        this.logger.log(`Cleaned up ${deleted} processed outbox events`);
      }
    } catch (error) {
      this.logger.error('Failed to cleanup outbox', error);
    } finally {
      await this.redis.releaseLock(`${this.LOCK_KEY}:cleanup`, lockValue);
    }
  }
}
