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

/**
 * 凌晨兜底扫描 — 每日 02:00 跑。
 *
 * 目的:抓回主路径漏掉的"画像 stale"患者:
 *   - 死信队列没消化的(persona-recompute 重试耗尽 → 等次日补)
 *   - 应用错误期间漏入队的(部署 / panic 错过 enqueue)
 *   - cold-import 进来但没触发 enqueue 的(脚本路径)
 *
 * 判定(任一命中即 stale,或 persona 根本不存在)→ enqueue persona-recompute(jobId 去重):
 *   - max(patient_transactions.event_seq) > personas.event_watermark   事务水位(辅)
 *   - max(patient_facts.updated_at)       > personas.fact_watermark    事实水位(主)
 *   - personas.next_boundary_at          <= now()                      到期改档
 *
 * ⚠️ 后两条都是 2026-07 补的,各修一个洞:
 *    · fact 水位 —— 画像读 patient_facts,原来只判 transaction,reparse(重写事实、不产事务)
 *      后的重算被全部 noop 掉。存量 fact_watermark 为 NULL → 本扫描持续判 stale,自然回填。
 *    · 到期改档 —— 有一类变化不伴随任何事实变动(时间跨过档界)。存量 next_boundary_at 为
 *      NULL,而 NULL 的语义是"不再改档",**不会自愈**,必须部署后强制回填一次:
 *      `pnpm recompute-persona -- --host=<h> --concurrency=N --force`。
 *
 * 安全:扫描走 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> {
    // 写侧总闸(见 scheduler-switch):迁移/备用实例不重算 persona —— 它会 enqueue 重算任务、
    // 改业务数据,与旧实例并跑会让待迁移的快照持续偏移。
    if (schedulerDisabled()) {
      this.logger.warn('stale-scan: ⏸ PAC_SCHEDULER_DISABLED=1,跳过本轮(迁移/备用实例)');
      return;
    }
    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"的患者,两类各走一条子查询(UNION):
   *
   * ① 输入变了 —— 有 transaction 但没 persona(从来没算过)/ event 水位落后 /
   *    **fact 水位落后**(reparse、事实层订正这类不产新事务的变更)。
   * ② 到期改档 —— 事实一个字没变,但时间跨过了某条档界(急迫 30/90 天、生命周期
   *    180/540/730、RFM 的 R、年龄跨段、滑动窗口过期)。这类患者只能靠 next_boundary_at
   *    捞:它是**确切时刻**,所以每晚只命中几百到几千条,走 (superseded_at, next_boundary_at)
   *    索引,而不是把 40 万人全重算一遍 —— 那等于把水位闸拆了。
   *
   * 返回 (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))
      )
      UNION
      (
        -- ② 到期改档(走 (superseded_at, next_boundary_at) 索引,每晚只有一小撮)
        SELECT host_id, tenant_id, patient_id
        FROM personas
        WHERE superseded_at IS NULL
          AND next_boundary_at IS NOT NULL
          AND next_boundary_at <= now()
      )
      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 };
  }
}
