import { BadRequestException, Injectable, Logger } from '@nestjs/common';
import { ApiCode, CanonicalResourceMeta, SyncDirection, SyncStatus } from '@pac/types';
import type { CanonicalResourceKey } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error';
import { PrismaService } from '../../../prisma/prisma.service';
import {
  normalizeCanonical,
  normalizeDatetime,
  type NormalizeContext,
} from '../assembler/field-mapper';
import { PipelineDispatcher, type EmitsResolver } from '../pipeline/pipeline-dispatcher.service';
import { ColdImportService } from '../cold-import/cold-import.service';
import { QueueProducer } from '../../../queues/queue-producer.service';
import type { EmitsConfig } from '../assembler/assembler.schema';
import type { PushEvent, PushEventsResponse, PushRowsResponse } from './push.schema';

/**
 * PushReceiver — 处理已验签的 push 事件批。
 *
 * 流程:
 *   1. SyncLog start(direction=push, triggered_by='push:<host>:<first_source_event_id>')
 *   2. events → tables(按 subjectType 分组,把 payload + 元字段拍平为 canonical row)
 *   3. PipelineDispatcher 跑 dispatch
 *   4. SyncLog finalize + 返回 metrics
 *
 * **同步执行**:webhook 直接 ack metrics 给宿主。
 *   - 短期(目前)够用,体量 < 100 events/req
 *   - 后续接 BullMQ 异步化:本服务只做"验签 + 落 transaction + ack",parser 进队列
 */
@Injectable()
export class PushReceiverService {
  private readonly logger = new Logger(PushReceiverService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly dispatcher: PipelineDispatcher,
    private readonly coldImport: ColdImportService,
    private readonly queue: QueueProducer,
  ) {}

  /**
   * 形态 A:接收原生表行(/push/rows)。
   *
   * 复用 cold-import/reparse 同核(ColdImportService.ingestRawTables):transforms + assembler + 版本流落库,
   * 然后对 touched 患者入队【画像重算】(plan 不在此触发,由定时任务兜底)。
   */
  async receiveRows(input: {
    hostId: string;
    hostName: string;
    source: string;
    rows: Record<string, unknown>[];
    /// 预检(?dryRun=1):跑完整管线但不落库,也不触发重算。宿主对接自测用。
    dryRun?: boolean;
  }): Promise<PushRowsResponse> {
    const { hostId, hostName, source, rows } = input;
    const dryRun = !!input.dryRun;

    const r = await this.coldImport.ingestRawTables({
      hostName,
      source,
      rows,
      dryRun,
      triggeredBy: `${dryRun ? 'push-dryrun' : 'push'}:${hostName}:${source}`,
    });

    const agg = r.perResource.reduce(
      (a, s) => ({
        txn: a.txn + s.transactionsWritten,
        dup: a.dup + s.duplicates,
        failed: a.failed + s.failed,
        factsFailed: a.factsFailed + s.factsFailed,
        stubs: a.stubs + s.patientStubsCreated,
      }),
      { txn: 0, dup: 0, failed: 0, factsFailed: 0, stubs: 0 },
    );

    // fact.content 被 schema 拒收的明细 —— 必须透出去,否则宿主只能看到"成功但没进数据"。
    // 2026-07-29 事故:4 行负金额结算被拒,回执却是 success/failed=0,对方无从判断,
    // 只看到超时 → 重推同一批 13 小时。定位信息(哪条 subject、哪个字段)不含 content 原文。
    const factRejects = r.perResource.flatMap((s) => s.factRejects);
    if (factRejects.length > 0) {
      this.logger.warn(
        `push/rows host=${hostName} source=${source} 有 ${agg.factsFailed} 条 fact 被 schema 拒收,` +
          `样本=${JSON.stringify(factRejects.slice(0, 3))}`,
      );
    }

    // ── 关键字段守卫:整批全失败(非重复、无一落库)→ 拒收报错,别让宿主静默丢数据 ──
    // 典型成因:宿主改了身份/时间列名 → 逐行合成失败。失败行不落任何库(raw 只活在 txn),
    // 拒绝是无损的;宿主修好后整批重推即可(幂等)。SyncLog 已标 FAILED + 告警(ingestRawTables 内)。
    if (rows.length > 0 && agg.txn === 0 && agg.dup === 0 && agg.failed > 0) {
      throw new BizError(
        ApiCode.SYNC_CONTRACT_DRIFT,
        `整批 ${agg.failed}/${rows.length} 行全部失败 —— 疑似关键字段缺失/改名,请核对数据字典` +
          `(syncLogId=${r.syncLogId})`,
      );
    }

    // 事件驱动:touched 患者入队 persona-recompute(去抖:存量批次重放时窗内合并,见 QueueProducer)
    // ⚠️ plan 不在此触发(persona 完成后链式 enqueue),与单患者刷新路径有意区分。
    // 预检不产生任何副作用 → 不入队(touched 本就为空,此处显式短路,语义更清楚)。
    let personaEnqueued = 0;
    for (const { patientId, tenantId } of dryRun ? [] : r.touched) {
      try {
        await this.queue.enqueuePersonaRecompute(
          { hostId, tenantId, patientId, triggeredBy: `push:${hostName}` },
          { debounce: true },
        );
        personaEnqueued++;
      } catch (err) {
        this.logger.warn(
          `enqueue persona-recompute failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }

    this.logger.log(
      `push/rows host=${hostName} source=${source} rows=${rows.length} ` +
        `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} stubs=${agg.stubs} ` +
        `personaEnqueued=${personaEnqueued}`,
    );
    // 空壳 = 引用的患者号在 PAC 找不到。顺序无关本就靠它兜底,但持续 >0 说明患者号对不上
    // (id 格式漂移 / 主档漏推),会造出无名幽灵患者混进召回池 —— 必须让宿主看见。
    if (agg.stubs > 0) {
      this.logger.warn(
        `push/rows host=${hostName} source=${source} 新建了 ${agg.stubs} 个空壳患者主档 —— ` +
          `若非首次推送该表,请核对 patient_id 格式(常见:整数被 float 化成 "676600.0")`,
      );
    }

    // 预检:回样本 canonical(各资源首行),供宿主核对 PAC 把原生行翻译成了什么
    const samples = dryRun
      ? r.perResource
          .filter((s) => (s.sampleCanonical?.length ?? 0) > 0)
          .map((s) => ({ resource: s.resource, canonical: s.sampleCanonical[0] }))
      : [];

    return {
      syncLogId: r.syncLogId,
      source,
      accepted: rows.length,
      transactionsWritten: agg.txn,
      duplicates: agg.dup,
      failed: agg.failed,
      /// transaction 已落账、但衍生 fact 被 content schema 拒收的条数(0 = 全部正常)
      factsFailed: agg.factsFailed,
      /// 上面那些拒收的定位明细(封顶若干条),宿主据此定位到自己源库的具体行
      factRejects,
      personaEnqueued,
      /// 本批新建的空壳患者主档数(引用的患者号在 PAC 找不到)。持续 >0 = 患者号对不上
      patientStubsCreated: agg.stubs,
      mappingMisses: r.mappingMisses.length,
      suspectFields: r.suspectFields.length,
      dryRun,
      samples,
    };
  }

  async receive(input: {
    hostId: string;
    hostName: string;
    events: PushEvent[];
  }): Promise<PushEventsResponse> {
    const { hostId, hostName, events } = input;

    // 集团 = 事件携带的 tenantId(同登录授权口径);一批要求同属一个集团。
    const tenantId = events[0]?.tenantId ?? '';
    if (!tenantId || events.some((e) => e.tenantId !== tenantId)) {
      throw new BadRequestException(
        !tenantId ? 'push 事件缺 tenantId' : '同一批 push 事件必须同属一个集团(tenantId)',
      );
    }

    // host→tenant 一致性校验:tenantId 必须是该 host 已建立的集团之一。
    // 注:跨 host 隔离已由所有键的 host_id 保证(host 写不进别人数据);此处只防 host 自己
    // 填错 tenant slug 把数据散进新分区。新 host(尚无任何数据)首次放行,由本次写入建立其集团
    // —— 存量可全程走 push(按批串行重放,见 channel-push 文档"存量接入"),不再要求先 cold-import。
    const knownTenants = await this.prisma.patient.groupBy({
      by: ['tenantId'],
      where: { hostId },
    });
    if (knownTenants.length > 0 && !knownTenants.some((k) => k.tenantId === tenantId)) {
      throw new BadRequestException(
        `tenantId='${tenantId}' 不是 host=${hostName} 已建立的集团(已知 ` +
          `${knownTenants.map((k) => k.tenantId).join(' / ')});新集团请先经 cold-import / pull 建立`,
      );
    }

    const triggeredBy = `push:${hostName}:${events[0]?.subjectId ?? 'unknown'}`;

    // ─── 1. SyncLog start ───
    const syncLog = await this.prisma.syncLog.create({
      data: {
        hostId,
        tenantId,
        direction: SyncDirection.PUSH,
        resource: 'push_batch',
        triggeredBy,
        status: SyncStatus.RUNNING,
      },
    });

    // ─── 2. events → tables(按 subjectType 分组 + Stage② 归一)───
    // 归一与形态 A / pull 走同一个 normalizeCanonical(消除同源分叉):金额(元→分)/
    // 无时区时间(补 offset)/ 布尔 coerce,按全局 CanonicalResourceMeta 识别列。
    // 配置来自 manifest(纯 C 宿主没建 → 默认 分 + Asia/Shanghai,即原契约语义,向后兼容)。
    const normCtx: NormalizeContext = this.coldImport.getNormalizeContext(hostName) ?? {
      amountUnit: 'fen',
      timezone: 'Asia/Shanghai',
    };
    const tables = this.eventsToTables(events, normCtx);

    // ─── 3. dispatch(emits resolver:per-row,从行的 _action / _subjectType 元字段读) ───
    let metrics;
    let firstError: string | null = null;
    try {
      metrics = await this.dispatcher.dispatchTables({
        tables,
        hostId,
        tenantId,
        sourceContext: `push:${hostName}`,
        syncLogId: syncLog.id,
        emitsResolver: this.buildEmitsResolver(),
        personaDebounce: true, // push 摄入统一去抖(存量批次重放时窗内合并)
      });
    } catch (err) {
      firstError = err instanceof Error ? err.message : String(err);
      metrics = {
        patientsUpserted: 0,
        transactionsWritten: 0,
        duplicates: 0,
        failed: events.length,
        factsCreated: 0,
        factsSuperseded: 0,
        factsUnchanged: 0,
        factsFailed: 0,
      };
    }

    // ─── 4. finalize ───
    const status =
      firstError && metrics.transactionsWritten === 0
        ? SyncStatus.FAILED
        : firstError
          ? SyncStatus.PARTIAL
          : SyncStatus.SUCCESS;

    await this.prisma.syncLog.update({
      where: { id: syncLog.id },
      data: {
        status,
        fetched: events.length,
        transactionsWritten: metrics.transactionsWritten,
        factsEmitted: metrics.factsCreated + metrics.factsSuperseded,
        duplicates: metrics.duplicates,
        failed: metrics.failed,
        errorMessage: firstError,
        endedAt: new Date(),
      },
    });

    this.logger.log(
      `push received host=${hostName} events=${events.length} status=${status} ` +
        `txn=${metrics.transactionsWritten} dups=${metrics.duplicates} failed=${metrics.failed}`,
    );

    return {
      syncLogId: syncLog.id,
      accepted: events.length,
      transactionsWritten: metrics.transactionsWritten,
      duplicates: metrics.duplicates,
      failed: metrics.failed,
    };
  }

  /**
   * 把 events[] 转换成 PipelineDispatcher 需要的 tables 形态,并做 Stage② 归一。
   *
   * 约定:event.subjectType 决定 table key(多元化:patient → patients,appointment → appointments)。
   * 每个 canonical row 由 event.payload + 元字段(externalId / clinicId / occurredAt 等)合成。
   * 归一按**单数** subjectType 查 CanonicalResourceMeta(表 key 是复数,meta key 是单数,别搞混);
   * 未知 subjectType 不归一,交后续管线处理。
   */
  private eventsToTables(events: PushEvent[], normCtx: NormalizeContext): Record<string, unknown[]> {
    const tables: Record<string, unknown[]> = {};

    for (const e of events) {
      const tableKey = pluralizeSubject(e.subjectType);
      if (!tables[tableKey]) tables[tableKey] = [];

      // canonical row = payload + 元字段拍平
      // envelope 时间字段(occurredAt/updatedAt)与 payload 同一转:无时区按配置补 offset
      //(synthesizer 的 new Date() / 乱序裁决的 parseSourceUpdatedAt 才不会按服务器时区误解)
      let canonicalRow: Record<string, unknown> = {
        ...e.payload,
        externalId: e.subjectId,
        clinicId: e.clinicId,
        occurredAt: normalizeDatetime(e.occurredAt, normCtx.timezone),
        // 源组织命名空间来自事件信封 → dispatcher 优先取它(防跨命名空间撞号)
        sourceUnit: e.sourceUnit ?? '',
        ...(e.patientId ? { patientExternalId: e.patientId } : {}),
        // 末次修改时间 → synthesizer 合幂等键(内容变升版本)+ 乱序裁决
        ...(e.updatedAt ? { updatedAt: normalizeDatetime(e.updatedAt, normCtx.timezone) } : {}),
        // 给 synthesizer / dispatcher 用的元字段
        _action: e.action,
        _subjectType: e.subjectType,
        // ⭐ 真原文留底:host event payload(不含 PAC 加的 meta 字段)
        __source_row: e.payload,
      };
      if (e.subjectType in CanonicalResourceMeta) {
        canonicalRow = normalizeCanonical(
          canonicalRow,
          e.subjectType as CanonicalResourceKey,
          normCtx,
        );
      }
      tables[tableKey]!.push(canonicalRow);
    }

    return tables;
  }

  /**
   * emits resolver — push 路径 per-row 提取 emits(同 batch 不同 event 可能 action 不同)。
   *
   * 从 canonicalRow 元字段读:
   *   - _action       → emits.action
   *   - _subjectType  → emits.subjectType
   *   - occurredAtField 固定 = 'occurredAt'(push 协议 canonical 字段名)
   *
   * patient 主档(resource='patients')返回 undefined,走 dispatcher.upsertPatient 路径。
   */
  private buildEmitsResolver(): EmitsResolver {
    return (resource, row) => {
      if (resource === 'patients') return undefined;
      const action = (row._action as string | undefined) ?? '';
      const subjectType = (row._subjectType as string | undefined) ?? '';
      if (!action || !subjectType) {
        this.logger.warn(
          `push event 缺 _action / _subjectType 元字段;resource=${resource},跳过`,
        );
        return undefined;
      }
      return {
        action,
        subjectType,
        occurredAtField: 'occurredAt',
      } as EmitsConfig;
    };
  }
}

/**
 * subjectType('patient' / 'appointment') → tables key('patients' / 'appointments').
 * 对应 CanonicalResourceKey。
 */
function pluralizeSubject(subjectType: string): string {
  if (subjectType.endsWith('s')) return subjectType;  // 已是复数
  return subjectType + 's';
}
