import { Injectable, Logger } from '@nestjs/common';
import type { Action } from '@pac/types';
import type { Prisma } from '@prisma/client';
import { PrismaService } from '../../../prisma/prisma.service';
import { QueueProducer } from '../../../queues/queue-producer.service';
import { TransactionSynthesizer } from './transaction-synthesizer';
import { ParserPipeline } from './parser-pipeline.service';
import type { EmitsConfig } from '../assembler/assembler.schema';
import { mergePatientPreferences } from './patient-preferences.util';

/** 患者索引 key:集团内 external_id 跨 source_unit 撞号,按 (source_unit, external_id) 建键。 */
function patientIndexKey(sourceUnit: string, externalId: string): string {
  return `${sourceUnit}#${externalId}`;
}
/** 从一行(canonical 或 raw)按命名空间字段取 source_unit;字段未声明 → ''(单命名空间)。 */
function resolveSourceUnit(row: Record<string, unknown>, field?: string): string {
  if (!field) return '';
  const v = row[field];
  return v === undefined || v === null ? '' : String(v).trim();
}

/**
 * PipelineDispatcher — "canonical tables → transaction + fact" 的共享内核。
 *
 * 三入口共用(canonical-fact-layer.md §五 4 入口统一管道):
 *   - PullOrchestrator(周期 cron 拉,emits 来自 yaml AssemblerEngine 推断)
 *   - ReconcileOrchestrator(对账,partial UNIQUE 自动去重)
 *   - PushReceiver(events,emits 来自 event 元字段)
 *
 * 注:不负责"raw → canonical"映射(那是 AssemblerEngine 的活,只有 HTTP adapter 需要);
 *      本服务输入**已 canonical 化**的 rows。
 *
 * 输入 tables 形如:
 *   {
 *     patients:     [{externalId, name, phone, ...}, ...],   // 主档,先做(不需 emits)
 *     appointments: [{externalId, patientExternalId, ...}],  // subject 资源(需 emits)
 *     encounters:   [...],
 *     ...
 *   }
 *
 * **emitsResolver**(v2.2,统一 4 入口契约):caller 必须提供"每行 → emits"的解析函数。
 *   - cold/pull(yaml 路径):resolver 按 resource 名查 yaml 注册表(同 resource 内所有行共用 emits)
 *   - push 路径:resolver 读 row._action / row._subjectType(同 resource 内每行可能 action 不同)
 *   - patient 主档:返回 undefined(走 upsert 不进 transaction)
 */
@Injectable()
export class PipelineDispatcher {
  private readonly logger = new Logger(PipelineDispatcher.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly synthesizer: TransactionSynthesizer,
    private readonly parserPipeline: ParserPipeline,
    private readonly queue: QueueProducer,
  ) {}

  async dispatchTables(input: {
    tables: Record<string, unknown[]>;
    hostId: string;
    tenantId: string;
    /** 患者命名空间字段(集团模型,= manifest.identity_namespace_field;如 'brand')。
     *  不传 = 单命名空间 host,source_unit=''。多品牌 host(jvs-dw reconcile)必传,否则撞号。 */
    identityNamespaceField?: string;
    /** 用于合成 source_event_id;**跨 run 稳定**否则永远不重复 */
    sourceContext: string;
    syncLogId?: string;
    /**
     * 解析每行的 emits 元数据(action/subjectType/occurredAtField)。
     *
     * - cold/pull yaml 路径:resolver 按 resource 名查 yaml 注册表
     * - push 路径:resolver 读 row._action / row._subjectType(per-row,因为同 batch 不同 action)
     * - patient 主档:返回 undefined(走 upsert 不进 transaction)
     */
    emitsResolver: EmitsResolver;
  }): Promise<DispatchMetrics> {
    const m = zeroMetrics();
    /** 本批触碰过的 patientId 集合,batch 末统一 enqueue persona-recompute(去重) */
    const touchedPatients = new Set<string>();

    // ─── 1. patients 先(建立 externalId → patientId 索引) ───
    const patientRows = (input.tables.patients ?? []) as Record<string, unknown>[];
    for (const row of patientRows) {
      try {
        const sourceUnit = resolveSourceUnit(row, input.identityNamespaceField);
        const patient = await this.upsertPatient(row, input.hostId, input.tenantId, sourceUnit);
        m.patientsUpserted++;
        // 患者主档变更也算"事件",触发 persona 重算(do_not_contact_status 等画像依赖它)
        if (patient?.id) touchedPatients.add(patient.id);
      } catch (err) {
        m.failed++;
        this.logger.error(
          `patient upsert failed externalId=${row.externalId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }

    const patientIndex = await this.buildPatientIndex(input.hostId, input.tenantId);

    // ─── 2. 其他 subject 资源:synthesize transaction + parser ───
    for (const [resource, rows] of Object.entries(input.tables)) {
      if (resource === 'patients') continue;
      for (const rawCanonicalRow of rows as Record<string, unknown>[]) {
        // v2.2 — emits 由 caller 提供的 resolver 决定:
        //   - cold/pull yaml 路径:emits 来自 yaml AssemblerEngine config
        //   - push 路径:emits 来自 event.action/event.subjectType(per-row)
        const emits = input.emitsResolver(resource, rawCanonicalRow);
        if (!emits) {
          this.logger.warn(
            `resource=${resource} 未提供 emits — emitsResolver 返回 undefined,跳过 transaction 合成`,
          );
          m.failed++;
          continue;
        }
        // ⭐ 提取真原文 link(W3 末加,合规留底):
        // - push: __source_row = event.payload(host event 原报文)
        // - mock pull: __source_row 未注入 → fallback 用 row 自身
        // 提取后从 canonical 里 strip,避免 __source_row 流入 transaction.canonical_payload
        const { __source_row: explicitSource, ...canonicalRow } = rawCanonicalRow as {
          __source_row?: Record<string, unknown>;
        } & Record<string, unknown>;
        const rawRow = explicitSource ?? canonicalRow;
        const sourceUnit = resolveSourceUnit(rawRow, input.identityNamespaceField);
        const txn = this.synthesizer.synthesize({
          rawRow,
          canonicalRow,
          emits,
          resource,
          hostId: input.hostId,
          tenantId: input.tenantId,
          patientIdResolver: (extId) => patientIndex.get(patientIndexKey(sourceUnit, extId)),
          sourceContext: input.sourceContext,
          sourceUnit,
        });
        if (!txn) {
          m.failed++;
          continue;
        }
        if (input.syncLogId) txn.syncLogId = input.syncLogId;

        try {
          const created = await this.prisma.patientTransaction.create({ data: txn });
          m.transactionsWritten++;
          await this.runParser(created, canonicalRow, m);
          if (created.patientId) touchedPatients.add(created.patientId);
        } catch (err) {
          const code = (err as { code?: string }).code;
          if (code === 'P2002') {
            m.duplicates++;
            // 幂等命中:仍跑 parser(上次 txn 落了 fact 没落的恢复场景)
            const existing = await this.prisma.patientTransaction.findFirst({
              where: {
                hostId: txn.hostId,
                tenantId: txn.tenantId,
                sourceEventId: txn.sourceEventId ?? undefined,
              },
            });
            if (existing) {
              await this.runParser(existing, canonicalRow, m);
              if (existing.patientId) touchedPatients.add(existing.patientId);
            }
          } else {
            m.failed++;
            this.logger.error(
              `txn write failed resource=${resource} subject=${txn.subjectId}: ${err instanceof Error ? err.message : err}`,
            );
          }
        }
      }
    }

    // ─── 3. 事件驱动:本批涉及的 patient 全部 enqueue persona-recompute ───
    // BullMQ jobId 去重:同 patient 重复 enqueue 只产生 1 个 waiting job
    // PersonaService.recompute 内部还有 eventWatermark 幂等兜底
    if (touchedPatients.size > 0) {
      m.personaEnqueued = touchedPatients.size;
      for (const patientId of touchedPatients) {
        try {
          await this.queue.enqueuePersonaRecompute({
            hostId: input.hostId,
            tenantId: input.tenantId,
            patientId,
            triggeredBy: input.sourceContext.startsWith('txn:')
              ? input.sourceContext
              : `txn:${input.sourceContext}`,
          });
        } catch (err) {
          // 入队失败不破坏主流程,只 warn(凌晨 cron 兜底会扫到 stale persona)
          this.logger.warn(
            `enqueue persona-recompute failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
          );
        }
      }
    }

    return m;
  }

  private async runParser(
    txn: { id: string; hostId: string; tenantId: string; patientId: string | null; action: string; subjectType: string; subjectId: string; occurredAt: Date; clinicId: string },
    canonicalRow: Record<string, unknown>,
    m: DispatchMetrics,
  ): Promise<void> {
    // patientId 可能 null(push 首见 stub 患者尚未补全) → 跳过 parser,等下次 pull 补全后回放
    if (!txn.patientId) {
      this.logger.debug(
        `skip parser (patientId=null) subject=${txn.subjectId} action=${txn.action}`,
      );
      return;
    }
    const r = await this.parserPipeline.runForTransaction({
      transaction: {
        id: txn.id,
        hostId: txn.hostId,
        tenantId: txn.tenantId,
        patientId: txn.patientId,
        action: txn.action as Action,
        subjectType: txn.subjectType,
        subjectId: txn.subjectId,
        occurredAt: txn.occurredAt,
        clinicId: txn.clinicId,
      },
      canonicalRow,
    });
    m.factsCreated += r.factsCreated;
    m.factsSuperseded += r.factsSuperseded;
    m.factsUnchanged += r.factsUnchanged;
    m.factsFailed += r.factsFailed;
  }

  private async upsertPatient(
    row: Record<string, unknown>,
    hostId: string,
    tenantId: string,
    sourceUnit: string,
  ): Promise<{ id: string }> {
    const externalId = String(row.externalId ?? '').trim();
    if (!externalId) throw new Error('patient row 缺 externalId');

    const patientData = {
      name: (row.name as string | undefined) ?? null,
      phone: (row.phone as string | undefined) ?? null,
      gender: (row.gender as string | undefined) ?? null,
      birthDate: row.birthDate ? new Date(row.birthDate as string) : null,
      // 专属客服(current_task_director)并入 preferences.dedicatedCs(零迁移)
      preferences: mergePatientPreferences(row),
      active: ((row.status as string | undefined) ?? 'active') === 'active',
    };
    const profileData = {
      doNotContact: (row.doNotContact as boolean | undefined) ?? false,
      deceased: (row.deceased as boolean | undefined) ?? false,
      tags: Array.isArray(row.tags) ? (row.tags as string[]) : [],
      notes: (row.notes as string | undefined) ?? null,
      // 获客渠道(A.2.1):一级 PAC 标准枚举 / 二级 host 原值(assembler 已映射)
      acquisitionChannel: (row.acquisitionChannel as string | undefined) ?? null,
      acquisitionSub: (row.acquisitionSub as string | undefined) ?? null,
      // 转介绍(B.1.2):recommend_num / recommend_amount(元→分)
      referralCount: row.referralCount != null ? Math.trunc(Number(row.referralCount)) : null,
      referralAmountCents:
        row.referralAmount != null ? Math.round(Number(row.referralAmount) * 100) : null,
    };

    const patient = await this.prisma.patient.upsert({
      where: { hostId_tenantId_sourceUnit_externalId: { hostId, tenantId, sourceUnit, externalId } },
      create: { hostId, tenantId, sourceUnit, externalId, ...patientData },
      update: patientData,
    });
    await this.prisma.patientProfile.upsert({
      where: { patientId: patient.id },
      create: { patientId: patient.id, ...profileData },
      update: profileData,
    });
    return { id: patient.id };
  }

  private async buildPatientIndex(
    hostId: string,
    tenantId: string,
  ): Promise<Map<string, string>> {
    const rows = await this.prisma.patient.findMany({
      where: { hostId, tenantId },
      select: { id: true, externalId: true, sourceUnit: true },
    });
    // 集团内 external_id 跨 source_unit 撞号 → 索引按 (source_unit, external_id) 建键
    return new Map(rows.map((r) => [patientIndexKey(r.sourceUnit, r.externalId), r.id]));
  }
}

/**
 * 解析每行的 emits 元数据 — caller 提供。
 * 返回 undefined → 该行跳过(不进 transaction);patient 主档行也用 undefined。
 */
export type EmitsResolver = (
  resource: string,
  row: Record<string, unknown>,
) => EmitsConfig | undefined;

export interface DispatchMetrics {
  patientsUpserted: number;
  transactionsWritten: number;
  duplicates: number;
  failed: number;
  factsCreated: number;
  factsSuperseded: number;
  factsUnchanged: number;
  factsFailed: number;
  /** 本批 enqueue 的 persona-recompute job 数(去重后,= 触碰过的 patient 数) */
  personaEnqueued: number;
}

function zeroMetrics(): DispatchMetrics {
  return {
    patientsUpserted: 0,
    transactionsWritten: 0,
    duplicates: 0,
    failed: 0,
    factsCreated: 0,
    factsSuperseded: 0,
    factsUnchanged: 0,
    factsFailed: 0,
    personaEnqueued: 0,
  };
}
