import { Injectable, Logger, NotFoundException } from '@nestjs/common';
// Prisma 作为值用(Prisma.DbNull:就地刷新时把 data 显式写回 SQL NULL)
import { Prisma } from '@prisma/client';
import { PrismaService } from '../../prisma/prisma.service';
import type { TenantScopeContext } from '../../common/decorators/tenant-scope.decorator';
import { FeatureRegistry } from './features/feature.registry';
import { PotentialTreatmentSelector } from '../clinical-gap/potential-treatment.selector';
import { personaWriteVerdict, type StoredPersonaFeature } from './persona-diff';
import type {
  ActiveFact,
  FeatureExtractorContext,
  PersonaFeatureDraft,
} from './features/feature.interface';

/** 画像输入水位:transaction(辅助)+ fact(主)。见 readWatermarks / isFresh。 */
interface Watermarks {
  eventSeq: bigint | null;
  factAt: Date | null;
}

/**
 * PersonaService — 第二层画像引擎
 *
 * 三个能力:
 *   1. recompute(patientId, source) — 跑所有 extractors,写新 Persona 版本(supersede 旧)
 *   2. getCurrent(patientId)        — 读当前 active 版本(含 features)
 *   3. getByVersion(patientId, v)   — 读历史版本
 *
 * 触发场景:
 *   - 事件驱动:transaction 落库后 BullMQ 队列 jobName=patient_id 串行调用(W3+)
 *   - 手动:CLI / Admin API
 *   - 定时兜底:每日 cron 全扫
 *
 * 水位:Persona.eventWatermark = 计算时所见 patient_transactions.eventSeq 最大值;
 *      下次重算前比对:max(event_seq) ≤ watermark → no-op(stale 判断)
 */
@Injectable()
export class PersonaService {
  private readonly logger = new Logger(PersonaService.name);

  /// 租户级 M(累计净消费)分位阈值缓存:key=host:tenant → { at, q:[p20,p40,p60,p80] }。
  /// RFM 的 M 打分要群体分位(图 B.1.1),但 PAC 按患者重算 → 分位阈值缓存复用(分布慢变);
  /// 批量重算每轮首个患者算一次,后续命中缓存;TTL 兜底单刷场景。
  private readonly mQuantileCache = new Map<string, { at: number; q: number[] }>();
  /** single-flight:缓存空/过期时,并发调用共享同一个在飞查询(防惊群)。 */
  private readonly mQuantileInflight = new Map<string, Promise<number[]>>();
  private static readonly M_QUANTILE_TTL_MS = 30 * 60 * 1000;

  constructor(
    private readonly prisma: PrismaService,
    private readonly registry: FeatureRegistry,
    private readonly gapSelector: PotentialTreatmentSelector,
  ) {}

  /**
   * 租户内"累计净消费/患者"的 [p20,p40,p60,p80](cents)。缓存 30min;失败/无数据 → []。
   *
   * 该查询全租户聚合(大库可达数秒)。**single-flight**:缓存空/过期时,并发调用(批量重算
   * concurrency>1)共享同一个在飞查询 —— 否则 N 个重查询同时打 DB 会卡死(370万 facts 实测)。
   */
  private async getMonetaryQuantiles(hostId: string, tenantId: string, nowMs: number): Promise<number[]> {
    const key = `${hostId}:${tenantId}`;
    const hit = this.mQuantileCache.get(key);
    if (hit && nowMs - hit.at < PersonaService.M_QUANTILE_TTL_MS) return hit.q;
    const inflight = this.mQuantileInflight.get(key);
    if (inflight) return inflight; // 已有查询在跑 → 复用,不再发一份

    const promise = (async () => {
      try {
        const rows = await this.prisma.$queryRaw<Array<{ q: number[] }>>`
          WITH spend AS (
            SELECT patient_id, SUM(
              CASE type WHEN 'payment_record'  THEN (content->>'amount_cents')::bigint
                        WHEN 'recharge_record' THEN (content->>'amount_cents')::bigint
                        WHEN 'refund_record'   THEN -(content->>'amount_cents')::bigint END) AS net
            FROM patient_facts
            WHERE host_id = ${hostId}::uuid AND tenant_id = ${tenantId}
              AND status IN ('active','fulfilled')
              AND type IN ('payment_record','recharge_record','refund_record')
            GROUP BY patient_id)
          SELECT percentile_cont(ARRAY[0.2,0.4,0.6,0.8]) WITHIN GROUP (ORDER BY net)::float8[] AS q FROM spend`;
        const q = (rows[0]?.q ?? []).map((x) => Number(x));
        this.mQuantileCache.set(key, { at: nowMs, q });
        return q;
      } catch (err) {
        this.logger.warn(`getMonetaryQuantiles failed (${key}): ${err instanceof Error ? err.message : err}`);
        return [];
      } finally {
        this.mQuantileInflight.delete(key);
      }
    })();
    this.mQuantileInflight.set(key, promise);
    return promise;
  }

  /**
   * 读该患者的两个输入水位 —— 一条 SQL 两个子查询,省一次往返。
   *
   * fact 侧刻意只按 patient_id 过滤:patient_id 全局唯一,加 host/tenant 反而让 planner
   * 用不上 (patient_id, updated_at) 索引(倒序扫首行即出,不回堆)。
   */
  private async readWatermarks(patient: {
    id: string;
    hostId: string;
    tenantId: string;
  }): Promise<Watermarks> {
    const rows = await this.prisma.$queryRaw<
      Array<{ event_seq: bigint | null; fact_at: Date | null }>
    >`
      SELECT
        (SELECT MAX(event_seq) FROM patient_transactions
          WHERE host_id = ${patient.hostId}::uuid AND tenant_id = ${patient.tenantId}
            AND patient_id = ${patient.id}::uuid) AS event_seq,
        (SELECT MAX(updated_at) FROM patient_facts
          WHERE patient_id = ${patient.id}::uuid) AS fact_at`;
    return { eventSeq: rows[0]?.event_seq ?? null, factAt: rows[0]?.fact_at ?? null };
  }

  /**
   * 当前 active 版本是否还"新鲜"(= 可以 noop)。
   *
   * ⚠️ **fact 水位是主,transaction 水位是辅**。历史上只判 transaction,而画像读的是
   * patient_facts —— reparse 重写事实但不产新事务,于是 2026-07-21 那次
   * 「amount_cents 应收→实付」reparse 后的全量重算被这道闸 noop 掉 325,979 人(生产日志),
   * 74% 存量画像的累计消费永久停在旧口径。fact 水位就是补这个洞。
   *
   * factWatermark 为 null = 存量老版本没记过 → 一律判 stale,靠下一次重算补上(自然回填)。
   * 两个水位都为 null = 该患者既无事务也无事实 → 没有输入可变,判新鲜(baseline 已写过)。
   */
  private isFresh(
    active: {
      eventWatermark: bigint | null;
      factWatermark: Date | null;
      nextBoundaryAt: Date | null;
    },
    wm: Watermarks,
    now: Date,
  ): boolean {
    if (wm.factAt !== null) {
      if (active.factWatermark === null) return false;
      if (wm.factAt.getTime() > active.factWatermark.getTime()) return false;
    }
    if (wm.eventSeq !== null) {
      if (active.eventWatermark === null) return false;
      if (wm.eventSeq > active.eventWatermark) return false;
    }
    // 到期改档:事实一个字没变,但时间跨过了某条档界(急迫 30/90 天、年龄跨段、滑窗过期…)。
    // 没这一条,夜扫把到期患者捞出来入队,也会被上面两个水位当场 noop 掉 —— 白捞。
    if (active.nextBoundaryAt !== null && active.nextBoundaryAt.getTime() <= now.getTime()) {
      return false;
    }
    return true;
  }

  /**
   * 重算单个 patient 的 Persona。
   *
   * 结果分五档(status):
   *   noop      闸拦下,压根没算(输入水位没动)
   *   unchanged 算了,内容一字未变 → 只推水位,不写 feature 行
   *   refreshed 只有易变数值变(天数/年龄/滑窗)→ 就地刷新,**不升版本**
   *   success   语义变了 → 产新版本并 supersede 旧 active(partial = 有 extractor 抛错)
   *   failed    全挂
   *
   * @returns 生效的 Persona id(unchanged/refreshed 时是原 active 的 id)
   */
  async recompute(input: {
    patientId: string;
    source: string; // 'txn:<id>' / 'manual:<userId>' / 'scheduled:<cronId>'
    now?: Date;
    force?: boolean; // 跳过水位幂等闸:算法/特征变更后重算(数据没变也要重跑)
  }): Promise<RecomputeResult> {
    const startedAt = new Date();
    const now = input.now ?? new Date();

    const patient = await this.prisma.patient.findUnique({
      where: { id: input.patientId },
      include: { profile: true },
    });
    if (!patient) {
      throw new NotFoundException(`Patient ${input.patientId} not found`);
    }

    // ─── 水位幂等闸(读侧省算)───
    // 自上次重算以来输入没变 → 直接 noop,省掉后面 ~6 次查询 + 16 个 extractor。
    // 水位有两个,fact 是主、transaction 是辅 —— 只盯 transaction 是历史事故根因,见 isFresh。
    const wm = await this.readWatermarks(patient);
    // 取最高版本行:正常流程下 supersede 与建新版同事务发生,所以它就是 active 版。
    // (旧代码这里查两次 —— 闸一次 where supersededAt:null、写入前又一次 orderBy version —— 已合并。)
    const latest = await this.prisma.persona.findFirst({
      where: { patientId: patient.id },
      orderBy: { version: 'desc' },
      select: {
        id: true,
        version: true,
        supersededAt: true,
        eventWatermark: true,
        factWatermark: true,
        nextBoundaryAt: true,
      },
    });
    const active = latest && !latest.supersededAt ? latest : null;

    if (!input.force && active && this.isFresh(active, wm, now)) {
      // 落一条 noop 日志便于审计 — 区别于实际算了一遍
      await this.prisma.personaRecomputeLog.create({
        data: {
          hostId: patient.hostId,
          tenantId: patient.tenantId,
          patientId: patient.id,
          triggeredBy: input.source,
          status: 'noop',
          producedPersonaId: active.id,
          featureCount: 0,
          ruleCount: 0,
          agentCount: 0,
          fallbackCount: 0,
          eventWatermark: wm.eventSeq,
          endedAt: new Date(),
        },
      });
      return {
        personaId: active.id,
        version: active.version,
        featureCount: 0,
        ruleCount: 0,
        failed: 0,
        status: 'noop',
        durationMs: Date.now() - startedAt.getTime(),
      };
    }

    // ⭐ 审计日志改「一次终态写」—— 原来是 create(running) + update(终态) 两次往返。
    //   startedAt 可显式赋值,所以耗时统计一字不差(见下方两处 create)。
    //   代价:进程被硬杀(OOM / SIGKILL)时这一条不会留痕。评估过可接受 ——
    //     · try/catch 里的异常仍会写 'failed' 行,真正的算法错误照样有记录;
    //     · 唯一丢的是"算到一半进程没了",而那种情况原本留下的是一条**永远卡在 running**
    //       的行(没人清理),本身就是噪声;批量回填另有 CLI 进度日志兜底。
    //   读侧确认无依赖:daily-health-report 只用 findFirst(最近一条 startedAt/status)
    //     与按 status 的 groupBy 日统计,都不需要 running 行。
    //   全量回填每患者省 1 次往返(11.7 → 10.7),且少一半日志表写入(35.6 万 → 17.8 万行/轮)。
    const logBase = {
      hostId: patient.hostId,
      tenantId: patient.tenantId,
      patientId: patient.id,
      triggeredBy: input.source,
      startedAt,
    };

    try {
      // ⭐ 一次取,内存分两份 —— 原来对 patient_facts 查两遍(同一患者、同一索引、两次往返):
      //     ① status IN (active, fulfilled)                    → factsByType
      //     ② type=appointment_record AND status != superseded → appointmentsAll
      //   ② 是 ① 的超集在 appointment 这一类上的放宽(多了 cancelled / no_show 等状态),
      //   所以取 `status != 'superseded'` 一把捞回来,再在内存里切成两份,语义逐字节等价。
      //   (全量回填时每患者省 1 次往返;12.7 → 11.7)
      const allFacts = await this.prisma.patientFact.findMany({
        where: {
          hostId: patient.hostId,
          tenantId: patient.tenantId,
          patientId: patient.id,
          status: { not: 'superseded' },
        },
        select: {
          id: true,
          subjectId: true,
          type: true,
          kind: true,
          status: true,
          occurredAt: true,
          plannedFor: true,
          clinicId: true,
          content: true,
        },
      });

      // factsByType 口径不变:只收 active / fulfilled。
      // 跟 chain-composer / scenario SQL 对齐:actual treatment 完成后 status='fulfilled',
      // 不能只取 'active' 否则已完成治疗在 persona 看不到 → treatment_chain_status 误算 gap
      const factsByType = new Map<string, ActiveFact[]>();
      // 全部 appointment(各状态,排 superseded)— 特别关注爽约/迟到用(no_show 预约 status≠active)
      const appointmentsAll: ActiveFact[] = [];
      for (const f of allFacts) {
        if (f.status === 'active' || f.status === 'fulfilled') {
          const arr = factsByType.get(f.type) ?? [];
          arr.push(f as ActiveFact);
          factsByType.set(f.type, arr);
        }
        if (f.type === 'appointment_record') appointmentsAll.push(f as ActiveFact);
      }

      // 关系边(家庭构成等用)— relatedPatientId 是否解析不影响,只看 relationship 类型
      const relations = await this.prisma.patientRelation.findMany({
        where: { patientId: patient.id },
        select: { relationship: true },
      });

      // 潜在治疗 gap(诊断了/建议了但没启动对应治疗)— 复用召回 gap 核心,去时间门(常态属性)。
      //   按患者 active 诊断/建议码剪枝:无相关码 → 不查 SQL。
      const activeCodes = new Set<string>();
      for (const t of ['diagnosis_record', 'recommendation_record']) {
        for (const f of factsByType.get(t) ?? []) {
          const code = (f.content as Record<string, unknown> | null)?.code;
          if (typeof code === 'string') activeCodes.add(code);
        }
      }
      const potentialGaps = activeCodes.size
        ? await this.gapSelector.selectForPatient({
            hostId: patient.hostId,
            tenantId: patient.tenantId,
            patientId: patient.id,
            now,
            activeCodes,
          })
        : [];

      // 水位在闸那里已经读过一次(wm),不再重查 —— 旧代码这条 findFirst 与闸里那条完全相同。
      const eventWatermark = wm.eventSeq;

      const ctx: FeatureExtractorContext = {
        patient: {
          id: patient.id,
          externalId: patient.externalId,
          name: patient.name,
          phone: patient.phone,
          gender: patient.gender,
          birthDate: patient.birthDate,
          active: patient.active,
        },
        profile: patient.profile
          ? {
              doNotContact: patient.profile.doNotContact,
              doNotContactReason: patient.profile.doNotContactReason,
              deceased: patient.profile.deceased,
              tags: patient.profile.tags,
              notes: patient.profile.notes,
              acquisitionChannel: patient.profile.acquisitionChannel,
              acquisitionSub: patient.profile.acquisitionSub,
              referralCount: patient.profile.referralCount,
              referralAmountCents: patient.profile.referralAmountCents,
            }
          : null,
        factsByType,
        relations,
        appointmentsAll,
        potentialGaps,
        now,
        populationStats: {
          monetaryQuantiles: await this.getMonetaryQuantiles(
            patient.hostId,
            patient.tenantId,
            now.getTime(),
          ),
        },
      };

      const drafts: PersonaFeatureDraft[] = [];
      let ruleCount = 0;
      let failed = 0;
      // 「到期改档」锚点:各特征报出自己因纯时间流逝而改档的最近时刻,取全体最小值。
      // 没有它,不依赖 fact 的特征(年龄段 / 年龄禁忌)永远等不到重算。
      let nextBoundaryAt: Date | null = null;
      for (const ex of this.registry.extractors) {
        try {
          const d = ex.extract(ctx);
          if (d) {
            drafts.push(d);
            ruleCount++;
          }
          const b = ex.nextBoundaryAt?.(ctx) ?? null;
          if (b && (!nextBoundaryAt || b < nextBoundaryAt)) nextBoundaryAt = b;
        } catch (err) {
          failed++;
          this.logger.error(
            `feature ${ex.key} extract failed for patient=${patient.id}: ${err instanceof Error ? err.message : err}`,
          );
        }
      }

      // ─── 写入判定(写侧省写)───
      // 闸放宽到 fact 水位之后,「算了但内容没变」会变多(reparse 常常只动了不进画像的字段)。
      // 这里决定:不写 / 就地刷新 / 升版本。三种都要推水位,否则闸永远开着、下次还白算一遍。
      const currentFeatures = active
        ? ((await this.prisma.personaFeature.findMany({
            where: { personaId: active.id },
            select: { key: true, description: true, score: true, data: true, evidence: true },
          })) as StoredPersonaFeature[])
        : null;
      const verdict = personaWriteVerdict(currentFeatures, drafts);
      const watermarks = { eventWatermark, factWatermark: wm.factAt, nextBoundaryAt };

      let personaId: string;
      let version: number;
      let status: string;

      if (verdict !== 'semantic' && active) {
        // 内容没变 / 只有天数漂移 → 不产新版本,认领与召回引用的 personaId 保持不变。
        personaId = active.id;
        version = active.version;
        status = verdict === 'unchanged' ? 'unchanged' : 'refreshed';
        await this.prisma.$transaction(
          async (tx) => {
            // 同下面建版本那支拿同一把患者级 advisory 锁:否则并发写者可能正在 supersede
            // 这一行 / 建新版本,我们的就地刷新就写到了一个刚变成历史的版本上。
            await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${patient.id}))`;
            if (verdict === 'volatile') {
              // key 集合与语义指纹都已确认相同 → 按 (personaId,key) 逐条就地改,保住行 id。
              for (const d of drafts) {
                await tx.personaFeature.update({
                  where: { personaId_key: { personaId: active.id, key: d.key } },
                  data: {
                    description: d.description,
                    score: d.score ?? null,
                    data:
                      d.data == null ? Prisma.DbNull : (d.data as unknown as Prisma.InputJsonValue),
                    evidence: d.evidence as unknown as Prisma.InputJsonValue,
                  },
                });
              }
            }
            await tx.persona.update({
              where: { id: active.id },
              // computedAt 不动(覆盖率日报靠它反推历史快照);就地刷新记 refreshedAt。
              data: { ...watermarks, ...(verdict === 'volatile' ? { refreshedAt: new Date() } : {}) },
            });
          },
          { maxWait: 30000, timeout: 60000 },
        );
      } else {
        // ⭐ 版本号必须在**锁内重读**,不能用事务外那次 findFirst 的结果。
        //   2026-07-27 测试服实测:CLI 全量回填跨过了 02:00 的 stale-scan cron,两个进程
        //   同时给同一患者建新版本 —— 各自在事务外读到同一个 latest.version、算出同一个
        //   nextVersion → `Unique constraint failed on (patient_id, version)`。
        //   那一轮 CLI 挂 688 个、定时任务自己挂 1,604 个。
        //   BullMQ 内部按 patient_id 串行挡得住自己人,挡不住外面另起的 CLI 进程。
        //   修法:进事务先拿**按患者的 advisory 事务锁**(随事务自动释放,跨进程有效),
        //   再重读当前最高版本 —— 迟到的那个会看到已经变大的版本号,顺序叠上去,不再冲突。
        let nextVersion = (latest?.version ?? 0) + 1;
        const created = await this.prisma.$transaction(
          async (tx) => {
            await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${patient.id}))`;
            // 锁内重读:并发写者可能已经把版本推高,也可能已经把我们看到的 active 换掉了
            const fresh = await tx.persona.findFirst({
              where: { patientId: patient.id },
              orderBy: { version: 'desc' },
              select: { id: true, version: true, supersededAt: true },
            });
            nextVersion = (fresh?.version ?? 0) + 1;
            const freshActive = fresh && !fresh.supersededAt ? fresh : null;
            if (freshActive) {
              await tx.persona.update({
                where: { id: freshActive.id },
                data: { supersededAt: new Date() },
              });
            }
            return tx.persona.create({
              data: {
                hostId: patient.hostId,
                tenantId: patient.tenantId,
                patientId: patient.id,
                version: nextVersion,
                source: input.source as unknown as string,
                ...watermarks,
                features: {
                  createMany: {
                    data: drafts.map((d) => ({
                      key: d.key,
                      description: d.description,
                      score: d.score ?? null,
                      data:
                        d.data == null
                          ? undefined // nullable Json 列省略 → NULL
                          : (d.data as unknown as Prisma.InputJsonValue),
                      evidence: d.evidence as unknown as Prisma.InputJsonValue,
                    })),
                  },
                },
              },
            });
          },
          { maxWait: 30000, timeout: 60000 },
        );
        personaId = created.id;
        version = nextVersion;
        status = failed === 0 ? 'success' : drafts.length > 0 ? 'partial' : 'failed';
      }

      await this.prisma.personaRecomputeLog.create({
        data: {
          ...logBase,
          status,
          producedPersonaId: personaId,
          featureCount: drafts.length,
          ruleCount,
          agentCount: 0,
          fallbackCount: 0,
          eventWatermark,
          endedAt: new Date(),
        },
      });

      return {
        personaId,
        version,
        featureCount: drafts.length,
        ruleCount,
        failed,
        status,
        durationMs: Date.now() - startedAt.getTime(),
      };
    } catch (err) {
      await this.prisma.personaRecomputeLog.create({
        data: {
          ...logBase,
          status: 'failed',
          errorMessage: err instanceof Error ? err.message : String(err),
          endedAt: new Date(),
        },
      });
      throw err;
    }
  }

  async getCurrent(scope: TenantScopeContext, patientId: string) {
    const p = await this.prisma.persona.findFirst({
      // 品牌隔离:persona 无 source_unit 列,经 patient 关系圈品牌(空=不限)。
      //   tenantId(集团)由 Prisma 纵深防御扩展自动注入。
      where: {
        patientId,
        supersededAt: null,
        ...(scope.sourceUnits.length ? { patient: { sourceUnit: { in: scope.sourceUnits } } } : {}),
      },
      include: { features: true },
      orderBy: { version: 'desc' },
    });
    return p ? this.serialize(p) : null;
  }

  private serialize(p: {
    id: string;
    hostId: string;
    tenantId: string;
    patientId: string;
    version: number;
    source: string;
    computedAt: Date;
    refreshedAt: Date | null;
    supersededAt: Date | null;
    eventWatermark: bigint | null;
    features: Array<{
      id: string;
      personaId: string;
      key: string;
      description: string;
      score: number | null;
      data: unknown;
      evidence: unknown;
      createdAt: Date;
    }>;
  }) {
    return {
      id: p.id,
      hostId: p.hostId,
      tenantId: p.tenantId,
      patientId: p.patientId,
      version: p.version,
      triggeredBy: p.source,
      computedAt: p.computedAt.toISOString(),
      // 就地刷新过就报刷新时刻 —— UI 的"更新于 N 天前"要反映内容真实新鲜度,
      // 而 computedAt 是版本创建时刻(覆盖率日报靠它反推历史快照,不能动)。
      refreshedAt: (p.refreshedAt ?? p.computedAt).toISOString(),
      supersededAt: p.supersededAt?.toISOString() ?? null,
      eventWatermark: p.eventWatermark != null ? p.eventWatermark.toString() : null,
      features: p.features.map((f) => ({
        id: f.id,
        personaId: f.personaId,
        key: f.key,
        description: f.description,
        score: f.score,
        data: (f.data as Record<string, unknown>) ?? null,
        evidence: (f.evidence as Record<string, unknown>) ?? {},
        createdAt: f.createdAt.toISOString(),
      })),
    };
  }
}

export interface RecomputeResult {
  personaId: string;
  version: number;
  featureCount: number;
  ruleCount: number;
  failed: number;
  status: string;
  durationMs: number;
}
