import { Injectable, Logger } from '@nestjs/common';
import { createHash } from 'node:crypto';
import { FactStatus } from '@pac/types';
import type { Prisma } from '@prisma/client';
import { PrismaService } from '../../../prisma/prisma.service';
import type { FactDraft } from './parsers/parser.interface';
import {
  FactContentSchemaError,
  validateFactContent,
} from './fact-content-schemas';

/** fact 版本链 key:集团内 subject_id 品牌级会撞,按 (source_unit, subject_id) 建键。 */
function factVersionKey(sourceUnit: string, subjectId: string): string {
  return `${sourceUnit}#${subjectId}`;
}

/** 时刻相等 — 双方皆缺(null/undefined)视为等;毫秒精度(列是 timestamptz(3))。 */
function sameInstant(a: Date | null | undefined, b: Date | null | undefined): boolean {
  return (a?.getTime() ?? null) === (b?.getTime() ?? null);
}

/**
 * FactWriter — patient_facts 版本流写入器
 *
 * 输入:一条 FactDraft + 触发它的 transactionId + (hostId, tenantId, patientId)
 *
 * 行为(对单 subject_id):
 *   1. 找当前 active 版本(partial UNIQUE 保证唯一)
 *   2. 计算 draft.content 的稳定 hash;比较 hash + status + 时间锚(occurredAt/plannedFor/validFrom/validUntil)
 *      ┌─ 全一致 → 幂等(把当前 transactionId 追加到 active.transactionIds,如未存在)
 *      └─ 任一不同 / 无 active → supersede 旧版本 → 插入新版本 version = old.version + 1(无则 1)
 *
 * 并发说明:
 *   - W2 是单进程顺序写,不需要锁
 *   - W3+ 上 BullMQ 时按 subject_id 串行化(jobName 对齐)+ 事务内 SELECT FOR UPDATE
 *   - schema 层 partial UNIQUE active 兜底
 *
 * 返回:WriteResult — 用于上层 metrics 统计(facts_created / facts_superseded / facts_unchanged)
 */
@Injectable()
export class FactWriter {
  private readonly logger = new Logger(FactWriter.name);

  constructor(private readonly prisma: PrismaService) {}

  async writeDraft(input: {
    draft: FactDraft;
    hostId: string;
    tenantId: string;
    /// 源组织命名空间(= 该患者 source_unit)。集团内 subject_id 品牌级会撞,进键/版本链消歧。
    sourceUnit: string;
    patientId: string;
    transactionId: string;
    /// 源记录末次修改时间(canonicalRow.updatedAt)。乱序防护:更旧的版本晚到 → 跳过不旧覆新。
    sourceUpdatedAt?: Date | null;
  }): Promise<FactWriteResult> {
    const { draft, hostId, tenantId, sourceUnit, patientId, transactionId, sourceUpdatedAt } = input;

    // ─── 闸 1:写库前 zod 强校验(canonical-fact-layer.md §二)──
    // 失败抛 FactContentSchemaError,由 ParserPipeline 计入 factsFailed 并 log issues。
    // 不允许"宽松落库"— 脏数据进 fact 层会污染整个下游(scenario / persona / AI)。
    let validatedContent: Prisma.InputJsonValue;
    try {
      validatedContent = validateFactContent(
        draft.type,
        draft.subjectId,
        draft.content,
      ) as Prisma.InputJsonValue;
    } catch (err) {
      if (err instanceof FactContentSchemaError) {
        this.logger.error(
          `[schema-violation] type=${err.factType} subjectId=${err.subjectId} ` +
            `transactionId=${transactionId} issues=${JSON.stringify(err.issues)}`,
        );
      }
      throw err;
    }

    const validatedDraft: FactDraft = { ...draft, content: validatedContent };
    const newHash = this.hashContent(validatedDraft.content);

    // 用一次 $transaction 把"读最新版本 → 决策 → supersede / insert"原子化。
    //
    // ⚠️ 必须查最新版本(不能只查 active)— 因为 fulfilled / cancelled / expired
    //    等终态版本"没有 active",但 (subjectId, version) UNIQUE 约束仍在 —
    //    要算下一 version 必须知道历史最高 version。
    // ⭐ 默认 timeout 5s 在 cold-import 大压力 / swap thrashing 下不够,撑到 60s
    //   fact-writer 是最热路径,每 fact 一次 tx,失败一次 = 整条 patient pipeline 中断
    return this.prisma.$transaction(async (tx) => {
      const latest = await tx.patientFact.findFirst({
        where: {
          hostId,
          tenantId,
          sourceUnit, // ⭐ 集团内 subject_id 跨品牌会撞,版本链必须按 source_unit 隔离,否则取错品牌版本
          subjectId: draft.subjectId,
        },
        orderBy: { version: 'desc' },
      });

      const draftStatus = draft.status ?? FactStatus.ACTIVE;

      // ── 乱序防护(业务时间裁决):双方都带 sourceUpdatedAt 且来的更旧 → 跳过,不旧覆新 ──
      // 场景:补推批快照(T 时刻读)晚于实时推送到达。同值(同秒改两次)不拦,交 hash 决策。
      // 任一方无 sourceUpdatedAt → 退回到达顺序语义(向后兼容)。
      if (
        latest?.sourceUpdatedAt &&
        sourceUpdatedAt &&
        sourceUpdatedAt.getTime() < latest.sourceUpdatedAt.getTime()
      ) {
        this.logger.debug(
          `stale skip subject=${draft.subjectId} incoming=${sourceUpdatedAt.toISOString()} < latest=${latest.sourceUpdatedAt.toISOString()}`,
        );
        return {
          action: 'stale_skipped',
          factId: latest.id,
          subjectId: draft.subjectId,
          version: latest.version,
        };
      }

      if (latest) {
        const latestHash = this.hashContent(latest.content as Prisma.InputJsonValue);
        const sameContent = latestHash === newHash;
        const sameStatus = latest.status === draftStatus;
        const sameTemporal = this.sameTemporalAnchors(latest, validatedDraft);

        if (sameContent && sameStatus && sameTemporal) {
          // 完全一致 — 幂等。把当前 transaction 加进证据数组(去重)
          if (!latest.transactionIds.includes(transactionId)) {
            await tx.patientFact.update({
              where: { id: latest.id },
              data: { transactionIds: [...latest.transactionIds, transactionId] },
            });
            return {
              action: 'evidence_appended',
              factId: latest.id,
              subjectId: draft.subjectId,
              version: latest.version,
            };
          }
          return {
            action: 'unchanged',
            factId: latest.id,
            subjectId: draft.subjectId,
            version: latest.version,
          };
        }

        // 有变化(content / status / 时间锚)— 如果 latest 是 active,supersede 它;
        // 终态版本(cancelled/fulfilled/expired/invalidated/superseded)不动,只追新版本。
        if (latest.status === FactStatus.ACTIVE) {
          await tx.patientFact.update({
            where: { id: latest.id },
            data: {
              status: FactStatus.SUPERSEDED,
              supersededAt: new Date(),
            },
          });
        }
      }

      const nextVersion = (latest?.version ?? 0) + 1;
      const created = await tx.patientFact.create({
        data: {
          hostId,
          tenantId,
          sourceUnit,
          patientId,
          subjectId: validatedDraft.subjectId,
          kind: validatedDraft.kind,
          type: validatedDraft.type,
          status: validatedDraft.status ?? FactStatus.ACTIVE,
          version: nextVersion,
          clinicId: validatedDraft.clinicId ?? null,
          occurredAt: validatedDraft.occurredAt ?? null,
          sourceUpdatedAt: sourceUpdatedAt ?? null,
          plannedFor: validatedDraft.plannedFor ?? null,
          validFrom: validatedDraft.validFrom ?? null,
          validUntil: validatedDraft.validUntil ?? null,
          title: validatedDraft.title ?? null,
          summary: validatedDraft.summary ?? null,
          content: validatedDraft.content,
          transactionIds: [transactionId],
        },
      });

      return {
        action: latest ? 'superseded' : 'created',
        factId: created.id,
        subjectId: draft.subjectId,
        version: nextVersion,
      };
    }, { maxWait: 30000, timeout: 60000 });
  }

  /**
   * 批量写入 N 个 fact draft(PR4 — 大幅减少 SQL 往返,9h→30min 性能优化的核心)。
   *
   * 流程:
   *   1. 全部 zod 校验(失败的整批抛错;调用方按 entry 级 try/catch 拆分小批重试)
   *   2. 收集所有 subject_id,**1 次 SELECT** 取全部 latest version
   *   3. 对每 entry 跟 latest 比 hash,决策:unchanged / evidence_append / supersede / create
   *   4. **3 个 bulk 操作** 在单个 $transaction:
   *      - updateMany supersede 那些 active 旧版本(by id IN ...)
   *      - 每个 evidence_append 单独 update(罕见路径,通常 0-N 个,代价小)
   *      - createMany 新版本(主流路径,百行级一次写完)
   *
   * 同 batch 同 subject_id 处理:
   *   - 若 N 个 draft 同 subject_id,按出现顺序处理 → 最后一个 draft 是最终 active 版本
   *   - 早 draft 跟 latest 比、决定 supersede;后续 draft 跟 早 draft 比、决定 supersede;...
   *   - 链式 supersede 在 1 个 batch 内完成,通过递推 latest 实现
   *
   * **跨 batch 的 subject_id**:不用担心(version + active partial UNIQUE 保证一致性)
   *
   * 失败处理:
   *   - 整批 $transaction 失败 → 全部回滚 → 抛 BulkWriteFailedError(调用方降级 per-entry)
   *   - **zod 失败 → 只踢掉违规的那几条**(rejected),其余照常走 bulk。
   *
   * ⚠️ zod 失败**曾经**是整批抛错的,2026-07-29 线上事故后改成隔离,别改回去:
   *   FRIDAY 推的 57,939 条结算里有 4 行负实收(退费的第三种表达,manifest 当时没覆盖),
   *   payment_record 的 amount_cents≥0 校验拒收 → 整批抛 → ParserPipeline 降级 per-entry →
   *   **每 fact 一个 $transaction** → 单批 200 行从 3 秒涨到 63.5 秒 → 超过宿主 60s 读超时 →
   *   对方收不到成功回执、无限重推同一批,实测卡了 13 小时一条没进。
   *   一行脏数据不该让同批另外 199 行陪葬 —— 隔离它、如实回报它,批照常走。
   */
  async bulkWrite(entries: BulkEntry[]): Promise<BulkWriteOutcome> {
    if (entries.length === 0) return { results: [], rejected: [] };

    // 假设全 batch 同 hostId+tenantId(processSubject 是按 host+tenant 跑的,符合)
    // 防御:校验所有 entry 同 host+tenant
    const { hostId, tenantId } = entries[0]!;
    for (const e of entries) {
      if (e.hostId !== hostId || e.tenantId !== tenantId) {
        throw new Error(
          `bulkWrite: 批内 hostId/tenantId 不一致 — host=${hostId} vs ${e.hostId}, tenant=${tenantId} vs ${e.tenantId}`,
        );
      }
    }

    // 1. 逐条 zod 校验 —— 违规的踢进 rejected,不连坐同批其余行(见上方 ⚠️)
    const validatedEntries: Array<BulkEntry & { validatedContent: Prisma.InputJsonValue; hash: string }> = [];
    const rejected: FactReject[] = [];
    for (const e of entries) {
      try {
        const content = validateFactContent(
          e.draft.type,
          e.draft.subjectId,
          e.draft.content,
        ) as Prisma.InputJsonValue;
        validatedEntries.push({ ...e, validatedContent: content, hash: this.hashContent(content) });
      } catch (err) {
        if (!(err instanceof FactContentSchemaError)) throw err; // 非 schema 错(编程错)不吞
        rejected.push({
          type: e.draft.type,
          subjectId: e.draft.subjectId,
          transactionId: e.transactionId,
          issues: err.issues.map((i) => `${i.path.join('.') || '(root)'}: ${i.message}`),
        });
        this.logger.error(
          `[schema-violation] type=${e.draft.type} subjectId=${e.draft.subjectId} ` +
            `transactionId=${e.transactionId} issues=${JSON.stringify(err.issues)}`,
        );
      }
    }
    if (validatedEntries.length === 0) return { results: [], rejected };

    // 2. 一次 SELECT 把所有相关 subject 的 latest version 拿回
    //    ⭐ 集团内 subject_id 跨品牌会撞 → 史表查带 source_unit 过滤,版本 map 按 (source_unit, subject_id) 建键
    const subjectIds = [...new Set(validatedEntries.map((e) => e.draft.subjectId))];
    const sourceUnits = [...new Set(validatedEntries.map((e) => e.sourceUnit))];
    const allHist = await this.prisma.patientFact.findMany({
      where: { hostId, tenantId, sourceUnit: { in: sourceUnits }, subjectId: { in: subjectIds } },
      orderBy: [{ subjectId: 'asc' }, { version: 'desc' }],
    });
    // 取每 (source_unit, subject) 的 highest version 行
    const latestBySubject = new Map<string, (typeof allHist)[number]>();
    for (const f of allHist) {
      const k = factVersionKey(f.sourceUnit, f.subjectId);
      if (!latestBySubject.has(k)) latestBySubject.set(k, f);
    }

    // 3. 决策每个 entry,累积批操作
    const toSupersede: string[] = []; // fact ids needing status -> superseded
    const toEvidenceAppend: Array<{ factId: string; transactionId: string }> = [];
    const toCreate: Prisma.PatientFactCreateManyInput[] = [];
    const results: FactWriteResult[] = [];
    // 模拟"链式 latest"— batch 内同 subject 多 draft 时,后 draft 跟前 draft 比
    const liveLatest = new Map<string, {
      id?: string;
      version: number;
      status: string;
      hash: string;
      transactionIds: string[];
      sourceUpdatedAt: Date | null;
      occurredAt: Date | null;
      plannedFor: Date | null;
      validFrom: Date | null;
      validUntil: Date | null;
    }>();
    for (const [k, latest] of latestBySubject.entries()) {
      liveLatest.set(k, {
        id: latest.id,
        version: latest.version,
        status: latest.status,
        hash: this.hashContent(latest.content as Prisma.InputJsonValue),
        transactionIds: latest.transactionIds,
        sourceUpdatedAt: latest.sourceUpdatedAt ?? null,
        occurredAt: latest.occurredAt ?? null,
        plannedFor: latest.plannedFor ?? null,
        validFrom: latest.validFrom ?? null,
        validUntil: latest.validUntil ?? null,
      });
    }

    for (const entry of validatedEntries) {
      const sid = entry.draft.subjectId;
      const vkey = factVersionKey(entry.sourceUnit, sid); // (source_unit, subject_id) 复合键
      const live = liveLatest.get(vkey);
      const draftStatus = entry.draft.status ?? FactStatus.ACTIVE;

      // 乱序防护(与 writeDraft 同语义):更旧的版本晚到 → 跳过,不旧覆新
      if (
        live?.sourceUpdatedAt &&
        entry.sourceUpdatedAt &&
        entry.sourceUpdatedAt.getTime() < live.sourceUpdatedAt.getTime()
      ) {
        this.logger.debug(
          `stale skip(bulk) subject=${sid} incoming=${entry.sourceUpdatedAt.toISOString()} < latest=${live.sourceUpdatedAt.toISOString()}`,
        );
        results.push({
          action: 'stale_skipped',
          factId: live.id ?? '',
          subjectId: sid,
          version: live.version,
        });
        continue;
      }

      if (
        live &&
        live.hash === entry.hash &&
        live.status === draftStatus &&
        this.sameTemporalAnchors(live, entry.draft)
      ) {
        // 内容一致 + 状态一致
        if (live.id && !live.transactionIds.includes(entry.transactionId)) {
          toEvidenceAppend.push({ factId: live.id, transactionId: entry.transactionId });
          // 更新 liveLatest.transactionIds(防同 subject 后续 draft 再加同一 tx)
          live.transactionIds = [...live.transactionIds, entry.transactionId];
          results.push({
            action: 'evidence_appended',
            factId: live.id,
            subjectId: sid,
            version: live.version,
          });
        } else {
          results.push({
            action: 'unchanged',
            factId: live.id ?? '',
            subjectId: sid,
            version: live.version,
          });
        }
        continue;
      }

      // 需要新版本(content / status 不同,或 batch 内同 subject 第二次出现)
      // supersede 旧 active(若有,且来自 DB,不是 batch 内"虚拟" liveLatest)
      if (live?.id && live.status === FactStatus.ACTIVE) {
        toSupersede.push(live.id);
      }
      const nextVersion = (live?.version ?? 0) + 1;
      toCreate.push({
        hostId,
        tenantId,
        sourceUnit: entry.sourceUnit,
        patientId: entry.patientId,
        subjectId: sid,
        kind: entry.draft.kind,
        type: entry.draft.type,
        status: draftStatus,
        version: nextVersion,
        clinicId: entry.draft.clinicId ?? null,
        occurredAt: entry.draft.occurredAt ?? null,
        sourceUpdatedAt: entry.sourceUpdatedAt ?? null,
        plannedFor: entry.draft.plannedFor ?? null,
        validFrom: entry.draft.validFrom ?? null,
        validUntil: entry.draft.validUntil ?? null,
        title: entry.draft.title ?? null,
        summary: entry.draft.summary ?? null,
        content: entry.validatedContent,
        transactionIds: [entry.transactionId],
      });
      // batch 内链式:把"虚拟 latest"刷新成本 draft 内容,后续同 subject 的 draft 用它当 prev
      // (注意 id=undefined 表示是本 batch 内 create,真正写入后才有 id;但 batch 内不会再 supersede 它,
      //  因为 batch 内同 subject 第二次出现也是 create,不 update)
      liveLatest.set(vkey, {
        id: undefined,
        version: nextVersion,
        status: draftStatus,
        hash: entry.hash,
        transactionIds: [entry.transactionId],
        sourceUpdatedAt: entry.sourceUpdatedAt ?? null,
        occurredAt: entry.draft.occurredAt ?? null,
        plannedFor: entry.draft.plannedFor ?? null,
        validFrom: entry.draft.validFrom ?? null,
        validUntil: entry.draft.validUntil ?? null,
      });
      results.push({
        action: live ? 'superseded' : 'created',
        factId: '', // bulk create 不返回 id;调用方不依赖 id(parser 不引用)
        subjectId: sid,
        version: nextVersion,
      });
    }

    // 4. 整批一个 transaction,真原子(supersede + create 不分离)
    // 用 callback 形态 + tx 重建 ops,这样可设 maxWait/timeout(默认 5s 在大批量 + swap 下不够)
    if (toSupersede.length > 0 || toEvidenceAppend.length > 0 || toCreate.length > 0) {
      await this.prisma.$transaction(
        async (tx) => {
          if (toSupersede.length > 0) {
            await tx.patientFact.updateMany({
              where: { id: { in: toSupersede } },
              data: { status: FactStatus.SUPERSEDED, supersededAt: new Date() },
            });
          }
          for (const e of toEvidenceAppend) {
            await tx.$executeRaw`
              UPDATE patient_facts
              SET transaction_ids = array_append(transaction_ids, ${e.transactionId}::uuid)
              WHERE id = ${e.factId}::uuid
                AND NOT (${e.transactionId}::uuid = ANY(transaction_ids))
            `;
          }
          if (toCreate.length > 0) {
            await tx.patientFact.createMany({ data: toCreate });
          }
        },
        { maxWait: 30_000, timeout: 120_000 },
      );
    }

    return { results, rejected };
  }

  /**
   * 时间锚相等判定 — 变更检测的组成部分(与 content hash / status 并列)。
   *
   * occurredAt / plannedFor / validFrom / validUntil 是事实实质:时间轴排序、
   * 召回窗口(COALESCE(occurred_at, planned_for))、过期 cron(valid_until)都直接消费。
   * 宿主纠正事件时间(如时区修复整体平移)时,content 往往不变 — 只比 hash+status
   * 会让纠正永远停在 evidence_appended,occurred_at 无法更新(时间锚不参与检测的旧 bug)。
   *
   * 不纳入:title / summary(展示层,规则引擎不读)、clinicId(如需纳入另行评估)。
   */
  private sameTemporalAnchors(
    latest: {
      occurredAt: Date | null;
      plannedFor: Date | null;
      validFrom: Date | null;
      validUntil: Date | null;
    },
    draft: Pick<FactDraft, 'occurredAt' | 'plannedFor' | 'validFrom' | 'validUntil'>,
  ): boolean {
    return (
      sameInstant(latest.occurredAt, draft.occurredAt) &&
      sameInstant(latest.plannedFor, draft.plannedFor) &&
      sameInstant(latest.validFrom, draft.validFrom) &&
      sameInstant(latest.validUntil, draft.validUntil)
    );
  }

  /**
   * 稳定 JSON hash — 递归按 key 排序后 sha256。
   * JSON.stringify 默认按 insertion order,key 顺序不同会算出不同 hash。
   */
  private hashContent(content: Prisma.InputJsonValue): string {
    return createHash('sha256').update(this.canonicalize(content)).digest('hex');
  }

  private canonicalize(v: unknown): string {
    if (v === null || v === undefined) return 'null';
    if (typeof v === 'number' || typeof v === 'boolean') return JSON.stringify(v);
    if (typeof v === 'string') return JSON.stringify(v);
    if (Array.isArray(v)) {
      return '[' + v.map((x) => this.canonicalize(x)).join(',') + ']';
    }
    if (typeof v === 'object') {
      const obj = v as Record<string, unknown>;
      const keys = Object.keys(obj).sort();
      return (
        '{' +
        keys
          .map((k) => JSON.stringify(k) + ':' + this.canonicalize(obj[k]))
          .join(',') +
        '}'
      );
    }
    return JSON.stringify(v);
  }
}

export interface FactWriteResult {
  /// created    — 新增首个版本(无前态)
  /// superseded — 旧 active 被置为 superseded,新版本插入
  /// evidence_appended — 内容未变,但本次 transaction 被加进证据数组
  /// unchanged  — 内容未变且 transaction 已在证据数组中(完全幂等)
  /// stale_skipped — 来的版本 sourceUpdatedAt 比库里更旧(乱序晚到)→ 忽略,不旧覆新
  action: 'created' | 'superseded' | 'evidence_appended' | 'unchanged' | 'stale_skipped';
  factId: string;
  subjectId: string;
  version: number;
}

/**
 * 被 fact.content schema 拒收的单条 draft —— 一路透到 push 回执,让宿主能自诊断。
 *
 * 只带定位信息(哪张源记录、哪个字段、为什么),**不带 content 原文**:
 * 回执会进对方日志,原文里有患者姓名/金额等,不该顺着回执外流。
 */
export interface FactReject {
  /// fact 类型(= assembler canonical,如 payment_record)
  type: string;
  /// 出问题的 subject(形如 payment_record:<宿主主键>)—— 宿主据此定位到具体那一行
  subjectId: string;
  transactionId: string;
  /// 人读的字段级原因,如 "amount_cents: Too small: expected number to be >=0"
  issues: string[];
}

/// bulkWrite 结果:成功写入的 + 被 schema 拒收隔离的(不抛错,不连坐)
export interface BulkWriteOutcome {
  results: FactWriteResult[];
  rejected: FactReject[];
}

/// FactWriter.bulkWrite 入参条目 — 一个 draft + 它所属的 patient / host / tenant / tx 上下文
export interface BulkEntry {
  draft: FactDraft;
  hostId: string;
  tenantId: string;
  /// 源组织命名空间(= 该患者 source_unit);集团内 subject_id 品牌级会撞,版本链按它隔离。
  sourceUnit: string;
  patientId: string;
  transactionId: string;
  /// 源记录末次修改时间(canonicalRow.updatedAt);乱序防护裁决用,null = 源没给
  sourceUpdatedAt?: Date | null;
}
