import { Injectable, Logger } from '@nestjs/common';
import { Cron } from '@nestjs/schedule';
import { PrismaService } from '../prisma/prisma.service';
import { QueueProducer } from './queue-producer.service';

/**
 * 凌晨兜底扫描 — 每日 02:00 跑。
 *
 * 目的:抓回主路径漏掉的"画像 stale"患者:
 *   - 死信队列没消化的(persona-recompute 重试耗尽 → 等次日补)
 *   - 应用错误期间漏入队的(部署 / panic 错过 enqueue)
 *   - cold-import 进来但没触发 enqueue 的(脚本路径)
 *
 * 判定(两个水位任一落后即算 stale,或 persona 根本不存在):
 *   - max(patient_transactions.event_seq) > personas.event_watermark
 *   - max(patient_facts.updated_at)       > personas.fact_watermark   ← 主水位
 *      → enqueue persona-recompute(BullMQ jobId 自动去重)
 *
 * ⚠️ fact 水位是 2026-07 补的:画像读 patient_facts,但原来只判 transaction,
 *    reparse(重写事实、不产事务)后的重算被全部 noop 掉。存量 persona 的 fact_watermark
 *    为 NULL → 本扫描会持续判它们 stale,以每晚 LIMIT 的速率**自然回填**。
 *    要一次性补完请跑 `pnpm recompute-persona -- --host=<h> --concurrency=N`。
 *
 * 安全:扫描走 GROUP BY + HAVING,不锁表。同一 patient 即便已在主路径队列里,
 *      jobId 去重保证不会重复算。
 */
@Injectable()
export class StaleScanService {
  private readonly logger = new Logger(StaleScanService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly queue: QueueProducer,
  ) {}

  /**
   * Stale persona 扫描 cron — env 驱动
   *
   * PAC_STALE_SCAN_CRON:
   *   不设         → 不跑(local 默认)
   *   '0 2 * * *'  → 生产 / staging 推荐(凌晨 02:00,DW 增量同步前一会儿)
   *
   * 量级估算:5 家试点 30 万患者,扫一次约 1-5 秒,enqueue 部分(预期 < 1% stale)。
   */
  @Cron(process.env.PAC_STALE_SCAN_CRON || '0 0 31 12 *' /* never — env not set */, {
    name: 'persona-stale-scan',
    // cron 表达式按北京时间解释(容器为 UTC;不加则实际 +8h)
    timeZone: 'Asia/Shanghai',
  })
  async scanAndEnqueueStale(): Promise<void> {
    const startedAt = Date.now();
    this.logger.log('stale-scan: START');

    const stale = await this.findStalePatients();

    let enqueued = 0;
    let failed = 0;
    for (const p of stale) {
      try {
        await this.queue.enqueuePersonaRecompute({
          hostId: p.hostId,
          tenantId: p.tenantId,
          patientId: p.patientId,
          triggeredBy: 'scheduled:stale-scan-2am',
        });
        enqueued++;
      } catch (err) {
        failed++;
        this.logger.warn(
          `stale-scan: enqueue failed patient=${p.patientId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }

    const durationMs = Date.now() - startedAt;
    this.logger.log(
      `stale-scan: DONE stale=${stale.length} enqueued=${enqueued} failed=${failed} duration=${durationMs}ms`,
    );
  }

  /**
   * 找出"画像 stale"的患者:
   *   - 有 transaction 但没 persona(从来没算过)
   *   - 有 persona 但 event 水位落后(漏了更新)
   *   - 有 persona 但 **fact 水位落后**(reparse / 事实层订正 —— 不产新事务的那类变更)
   *
   * 用一句 raw SQL 兼顾三种情况;返回 (hostId, tenantId, patientId) 三元组。
   */
  private async findStalePatients(): Promise<
    Array<{ hostId: string; tenantId: string; patientId: string }>
  > {
    const rows = await this.prisma.$queryRaw<
      Array<{ host_id: string; tenant_id: string; patient_id: string }>
    >`
      WITH tx AS (
        SELECT host_id, tenant_id, patient_id, MAX(event_seq) AS max_seq
        FROM patient_transactions
        WHERE patient_id IS NOT NULL
        GROUP BY host_id, tenant_id, patient_id
      ), fx AS (
        -- (patient_id, updated_at) 索引上的 index-only 聚合,不回堆
        SELECT patient_id, MAX(updated_at) AS max_fact_at
        FROM patient_facts
        GROUP BY patient_id
      )
      SELECT tx.host_id, tx.tenant_id, tx.patient_id
      FROM tx
      LEFT JOIN personas p
        ON p.patient_id = tx.patient_id
       AND p.superseded_at IS NULL
      LEFT JOIN fx ON fx.patient_id = tx.patient_id
      WHERE p.id IS NULL
         OR p.event_watermark IS NULL
         OR p.event_watermark < tx.max_seq
         OR (fx.max_fact_at IS NOT NULL
             AND (p.fact_watermark IS NULL OR p.fact_watermark < fx.max_fact_at))
      LIMIT 5000
    `;
    return rows.map((r) => ({
      hostId: r.host_id,
      tenantId: r.tenant_id,
      patientId: r.patient_id,
    }));
  }

  /**
   * 手动触发(管理后台 / CLI 用)。
   */
  async runNow(): Promise<{ stale: number; enqueued: number; failed: number }> {
    const stale = await this.findStalePatients();
    let enqueued = 0;
    let failed = 0;
    for (const p of stale) {
      try {
        await this.queue.enqueuePersonaRecompute({
          hostId: p.hostId,
          tenantId: p.tenantId,
          patientId: p.patientId,
          triggeredBy: 'manual:stale-scan-runNow',
        });
        enqueued++;
      } catch {
        failed++;
      }
    }
    return { stale: stale.length, enqueued, failed };
  }
}
