import { Injectable, Logger } from '@nestjs/common';
import * as fs from 'node:fs';
import * as path from 'node:path';
import * as yaml from 'js-yaml';
import { parse as parseCsv } from 'csv-parse/sync';
import { randomUUID } from 'node:crypto';
import { SyncDirection, SyncStatus } from '@pac/types';
import type { Action, CanonicalResourceKey } from '@pac/types';
import type { Prisma } from '@prisma/client';
import { PrismaService } from '../../../prisma/prisma.service';
import { TransactionSynthesizer } from '../pipeline/transaction-synthesizer';
import { ParserPipeline } from '../pipeline/parser-pipeline.service';
import { AssemblerEngine, type AssemblerResult } from '../assembler/assembler-engine';
import {
  AssemblerConfigSchema,
  type AssemblerConfig,
} from '../assembler/assembler.schema';
import {
  ColdImportManifest,
  ColdImportManifestSchema,
  inferFileFormat,
} from './manifest.schema';
import {
  ClickHouseSourceService,
  type IncrementalConfig,
  type PatientScope,
} from './clickhouse-source.service';
import { TransformEngine } from '../transforms/transform-engine';
import { buildTenantResolver, type TenantResolver } from './tenant-resolver';

/**
 * ColdImportService(v2 — AssemblerEngine 驱动)
 *
 * 流程:
 *   1. 读 manifest.yaml(声明 host 表 + assembler 配置)
 *   2. 读所有 raw table 文件到 `tables: Record<tableName, row[]>`
 *   3. 读所有 assembler yaml → 校验 zod
 *   4. 按 canonical='patient' 的 assembler 跑 patient 主档 upsert
 *   5. 其他 canonical 按依赖顺序跑 AssemblerEngine.assemble → 拿 canonical rows
 *      → TransactionSynthesizer 合成 transaction → 写库 → ParserPipeline 衍生 fact
 *   6. SyncLog 写账本(direction=pull,跟同套 pipeline 共用语义)
 *
 * host 工作量降至最低:`SELECT * FROM tb_xxx` 导出表 + 一份数据字典即可,
 * 不需要懂 PAC canonical / 不需要写 JSON 聚合 / 不需要翻译枚举(全 yaml 配置)。
 */
@Injectable()
export class ColdImportService {
  private readonly logger = new Logger(ColdImportService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly synthesizer: TransactionSynthesizer,
    private readonly parserPipeline: ParserPipeline,
    private readonly assembler: AssemblerEngine,
    private readonly chSource: ClickHouseSourceService,
    private readonly transformEngine: TransformEngine,
  ) {}

  async importDirectory(
    dir: string,
    options: { dryRun?: boolean; incremental?: boolean } = {},
  ): Promise<ImportRunResult> {
    const absDir = path.resolve(dir);
    const runId = randomUUID();
    this.logger.log(
      `Cold import starting: ${absDir} (dryRun=${!!options.dryRun}, runId=${runId})`,
    );

    // 1. Manifest
    const manifest = this.readManifest(absDir);

    // 2. Host
    const host = await this.prisma.host.findUnique({
      where: { name: manifest.host_name },
    });
    if (!host) {
      throw new Error(
        `Host not found by name=${manifest.host_name};请先 seed 或通过 admin API 创建`,
      );
    }
    // v2.3:per-row tenant resolver(支持单 host 多 tenant)
    const tenantResolver = buildTenantResolver(manifest);
    const knownTenants = tenantResolver.knownTenants();
    const seenTenants = new Set<string>(); // 跑过程动态收集(pass-through 模式不预知)
    this.logger.log(
      `Host=${host.name}(${host.id}) resolver=${tenantResolver.describe()}` +
        (knownTenants.length > 0 ? ` declared_tenants=[${knownTenants.join(',')}]` : ' (动态 pass-through)') +
        ` source=${manifest.sql_source ? 'clickhouse' : 'files'}` +
        ` assemblers=${manifest.assemblers.length}`,
    );

    // 3. 一次性加载所有 raw tables(文件 / ClickHouse 二选一)
    // W4 末:incremental 模式构建 cursor map(从最近一次成功 sync_log 的 cursor_after JSON 读)
    let incrementalConfig: IncrementalConfig | undefined;
    if (options.incremental) {
      const perQueryCfg = manifest.sql_source?.incremental?.per_query;
      if (!perQueryCfg) {
        throw new Error(
          `--incremental 模式但 manifest.sql_source.incremental.per_query 未配置;请在 manifest.yaml 补 cursor 配置`,
        );
      }
      const lastCursor = await this.readLastIncrementalCursor(host.id);
      this.logger.log(
        `Incremental cursor 读:${Object.keys(lastCursor).length === 0 ? '(首跑,全量)' : JSON.stringify(lastCursor)}`,
      );
      incrementalConfig = {
        perQuery: Object.fromEntries(
          Object.entries(perQueryCfg).map(([table, cfg]) => [
            table,
            { cursorColumn: cfg.cursor_column, cursorValue: lastCursor[table] ?? null },
          ]),
        ),
      };
    }
    let tables = await this.loadAllTables(absDir, manifest, incrementalConfig);

    // 3.5 Layer A.5 transforms — yaml 声明式形态改造(JSON 拆行 / 派生 / 路由等)
    if (manifest.transforms && manifest.transforms.length > 0) {
      this.logger.log(
        `Layer A.5 transforms:${manifest.transforms.length} 步,sources=${Object.keys(tables).length}`,
      );
      // cast 兼容:loadAllTables 返回 unknown[] per row,transforms 内部按 Record<string, unknown> 处理
      const transformInputs = tables as Record<string, Record<string, unknown>[]>;
      tables = this.transformEngine.run({ tables: transformInputs, transforms: manifest.transforms });
      this.logger.log(
        `Layer A.5 done,tables 现有 ${Object.keys(tables).length} 张` +
          ` (含 transform 产出):${Object.entries(tables)
            .map(([k, v]) => `${k}=${v.length}`)
            .join(', ')}`,
      );
    }

    // 4. 一次性读所有 assembler configs
    const assemblerConfigs = this.loadAllAssemblers(absDir, manifest);
    // 必须有 patient assembler(主档)
    const patientCfg = assemblerConfigs.find((c) => c.canonical === 'patient');
    const subjectCfgs = assemblerConfigs.filter((c) => c.canonical !== 'patient');

    // 5. SyncLog start
    // SyncLog 自身的 tenantId 列只能填一个;多 tenant 跑用 '_multi' sentinel
    const syncLogTenant = knownTenants.length === 1 ? knownTenants[0]! : '_multi';
    // W4 末:incremental 模式 → resource='incremental_bundle' + cursor_before 写入老 cursor
    const syncResource = options.incremental ? 'incremental_bundle' : 'cold_import_bundle';
    const cursorBeforeJson = options.incremental && incrementalConfig
      ? JSON.stringify(
          Object.fromEntries(
            Object.entries(incrementalConfig.perQuery).map(([k, v]) => [k, v.cursorValue ?? '']),
          ),
        )
      : null;
    const syncLog = options.dryRun
      ? null
      : await this.prisma.syncLog.create({
          data: {
            hostId: host.id,
            tenantId: syncLogTenant,
            direction: SyncDirection.PULL, // 冷启复用 pull 语义(同套 pipeline)
            resource: syncResource,
            triggeredBy: `${options.incremental ? 'incremental' : 'cold_import'}:${path.basename(absDir)}:${runId}`,
            status: SyncStatus.RUNNING,
            cursorBefore: cursorBeforeJson,
          },
        });

    // 6. 逐 assembler 处理
    const perResource: PerResourceStats[] = [];
    const totals = this.zeroTotals();
    let firstError: string | null = null;

    const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };

    // patients 先做(建立 patient_id 索引,供 subject 资源关联)
    if (patientCfg) {
      const stats = await this.processPatients(
        tables,
        patientCfg,
        host.id,
        tenantResolver,
        seenTenants,
        normalize,
        options.dryRun ?? false,
      );
      perResource.push(stats);
      totals.patientsUpserted += stats.patientsUpserted;
      totals.failed += stats.failed;
      if (!firstError && stats.failed > 0)
        firstError ||= `patient: ${stats.failed} rows failed`;
    } else {
      this.logger.warn(
        `No assembler for canonical='patient';skipping patient master upsert`,
      );
    }

    // 其他资源(transaction 合成 + parser 衍生 fact)
    for (const cfg of subjectCfgs) {
      try {
        const stats = await this.processSubject(
          tables,
          cfg,
          host.id,
          tenantResolver,
          seenTenants,
          normalize,
          syncLog?.id,
          options.dryRun ?? false,
        );
        perResource.push(stats);
        totals.transactionsWritten += stats.transactionsWritten;
        totals.duplicates += stats.duplicates;
        totals.failed += stats.failed;
        totals.factsCreated += stats.factsCreated;
        totals.factsSuperseded += stats.factsSuperseded;
        totals.factsUnchanged += stats.factsUnchanged;
        totals.factsEvidenceAppended += stats.factsEvidenceAppended;
        totals.factsFailed += stats.factsFailed;
      } catch (err) {
        const msg = err instanceof Error ? err.message : String(err);
        firstError ||= `${cfg.canonical}: ${msg}`;
        this.logger.error(`Resource ${cfg.canonical} failed: ${msg}`);
      }
    }

    // 7. SyncLog finalize
    const fetched =
      totals.patientsUpserted +
      totals.transactionsWritten +
      totals.duplicates +
      totals.failed;
    const status =
      totals.failed === 0
        ? SyncStatus.SUCCESS
        : totals.patientsUpserted + totals.transactionsWritten > 0
          ? SyncStatus.PARTIAL
          : SyncStatus.FAILED;

    // W4 末:incremental cursor_after — 合并老 cursor + 本批新增 max
    //   本批某表 0 行(无新数据)→ 保留老 cursor 不变(下次同样 WHERE > old_cursor)
    //   本批某表有行 → cursor 推进到本批 max(updated_date)
    let cursorAfterJson: string | null = null;
    if (options.incremental && incrementalConfig) {
      const oldCursors: Record<string, string> = JSON.parse(cursorBeforeJson ?? '{}');
      const advances = incrementalConfig.cursorAdvances ?? {};
      const merged = { ...oldCursors };
      for (const [tbl, newVal] of Object.entries(advances)) {
        const old = merged[tbl];
        if (!old || newVal > old) merged[tbl] = newVal;
      }
      cursorAfterJson = JSON.stringify(merged);
      this.logger.log(`Incremental cursor 写:${cursorAfterJson}`);
    }
    if (syncLog) {
      await this.prisma.syncLog.update({
        where: { id: syncLog.id },
        data: {
          status,
          fetched,
          transactionsWritten: totals.transactionsWritten,
          factsEmitted: totals.factsCreated + totals.factsSuperseded,
          duplicates: totals.duplicates,
          failed: totals.failed,
          errorMessage: firstError,
          endedAt: new Date(),
          cursorAfter: cursorAfterJson,
        },
      });
    }

    const result: ImportRunResult = {
      runId,
      hostId: host.id,
      hostName: host.name,
      // 实际跑过程见到的 tenant 集合(pass-through 模式下 declared 为空,这里是真实结果)
      tenantIds: [...seenTenants].sort(),
      dryRun: !!options.dryRun,
      syncLogId: syncLog?.id,
      status,
      totals,
      perResource,
    };

    this.logger.log(
      `Cold import done: status=${status} patients=${totals.patientsUpserted} txns=${totals.transactionsWritten} dups=${totals.duplicates} failed=${totals.failed} ` +
        `facts(created=${totals.factsCreated} superseded=${totals.factsSuperseded} unchanged=${totals.factsUnchanged} evidence=${totals.factsEvidenceAppended} failed=${totals.factsFailed})`,
    );
    return result;
  }

  // ─────────────────────────────────────────────────────────
  // W4 末:单患者按需刷新(详情页"刷新"按钮 / SyncService.runPullForPatient)
  //
  // 关键纪律(cursor 隔离):
  //   - resource='patient_refresh',跟 incremental_bundle / cold_import_bundle **分开 sync_logs**
  //   - 不读 sync_logs cursor(不被增量 cron 影响),**不写** cursor_after(不污染增量 cron)
  //   - daily cron 完全不受单患者刷新影响 ✓
  //
  // 等价 importDirectory({incremental:false}) 但 SQL 注入 patient_id+brand 精确范围:
  //   - chSource.loadTablesForPatient(scope) 改写 manifest queries 拉单患者全表
  //   - 同 transforms / assemblers / parser 链路(下游无感知)
  //   - source_event_id partial UNIQUE 自动幂等(多次刷新不重复)
  // ─────────────────────────────────────────────────────────
  async importPatient(
    dir: string,
    scope: PatientScope & { triggeredBy?: string },
  ): Promise<ImportRunResult> {
    const absDir = path.resolve(dir);
    const runId = randomUUID();
    this.logger.log(
      `Patient refresh starting: ${absDir} patient=${scope.patientExternalId} brand=${scope.brand} runId=${runId}`,
    );

    // 1. Manifest
    const manifest = this.readManifest(absDir);
    if (!manifest.sql_source) {
      throw new Error(
        `Patient refresh 需要 sql_source(ClickHouse 直连);manifest 当前是 tables[] 文件源,不支持`,
      );
    }

    // 2. Host
    const host = await this.prisma.host.findUnique({
      where: { name: manifest.host_name },
    });
    if (!host) {
      throw new Error(`Host not found by name=${manifest.host_name}`);
    }

    const tenantResolver = buildTenantResolver(manifest);
    const seenTenants = new Set<string>();

    // 3. 单患者 SQL — 改写所有 queries 加 WHERE patient_id+brand
    let tables = await this.chSource.loadTablesForPatient(manifest.sql_source, {
      patientExternalId: scope.patientExternalId,
      brand: scope.brand,
    });

    // 3.5 Layer A.5 transforms — 同 importDirectory(单患者数据量小,几十~几百行,毫秒级)
    if (manifest.transforms && manifest.transforms.length > 0) {
      const transformInputs = tables as Record<string, Record<string, unknown>[]>;
      tables = this.transformEngine.run({
        tables: transformInputs,
        transforms: manifest.transforms,
      });
    }

    // 4. Assemblers
    const assemblerConfigs = this.loadAllAssemblers(absDir, manifest);
    const patientCfg = assemblerConfigs.find((c) => c.canonical === 'patient');
    const subjectCfgs = assemblerConfigs.filter((c) => c.canonical !== 'patient');

    // 5. SyncLog — resource='patient_refresh' ⭐ 跟增量隔离 ⭐
    //    tenantId 用 resolver 已知的(单 brand → 唯一 tenant);若动态 pass-through 用 sentinel
    const knownTenants = tenantResolver.knownTenants();
    const syncLogTenant = knownTenants.length === 1 ? knownTenants[0]! : '_multi';
    const syncLog = await this.prisma.syncLog.create({
      data: {
        hostId: host.id,
        tenantId: syncLogTenant,
        direction: SyncDirection.PULL,
        resource: 'patient_refresh', // ⭐ 跟 incremental_bundle / cold_import_bundle 区分
        triggeredBy: scope.triggeredBy ?? `patient_refresh:${scope.patientExternalId}:${runId}`,
        status: SyncStatus.RUNNING,
        // cursorBefore / cursorAfter 都不写(本身不参与增量)
      },
    });

    // 6. 跑 patient 主档 + subjects(同 importDirectory 路径)
    const perResource: PerResourceStats[] = [];
    const totals = this.zeroTotals();
    let firstError: string | null = null;
    const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };

    if (patientCfg) {
      const stats = await this.processPatients(
        tables, patientCfg, host.id, tenantResolver, seenTenants, normalize, false,
      );
      perResource.push(stats);
      totals.patientsUpserted += stats.patientsUpserted;
      totals.failed += stats.failed;
      if (!firstError && stats.failed > 0) firstError ||= `patient: ${stats.failed} rows failed`;
    }

    for (const cfg of subjectCfgs) {
      try {
        const stats = await this.processSubject(
          tables, cfg, host.id, tenantResolver, seenTenants, normalize, syncLog.id, false,
        );
        perResource.push(stats);
        totals.transactionsWritten += stats.transactionsWritten;
        totals.duplicates += stats.duplicates;
        totals.failed += stats.failed;
        totals.factsCreated += stats.factsCreated;
        totals.factsSuperseded += stats.factsSuperseded;
        totals.factsUnchanged += stats.factsUnchanged;
        totals.factsEvidenceAppended += stats.factsEvidenceAppended;
        totals.factsFailed += stats.factsFailed;
      } catch (err) {
        const msg = err instanceof Error ? err.message : String(err);
        firstError ||= `${cfg.canonical}: ${msg}`;
        this.logger.error(`Patient refresh resource ${cfg.canonical} failed: ${msg}`);
      }
    }

    // 7. SyncLog finalize
    const fetched =
      totals.patientsUpserted + totals.transactionsWritten + totals.duplicates + totals.failed;
    const status =
      totals.failed === 0
        ? SyncStatus.SUCCESS
        : totals.patientsUpserted + totals.transactionsWritten > 0
          ? SyncStatus.PARTIAL
          : SyncStatus.FAILED;

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

    this.logger.log(
      `Patient refresh done: patient=${scope.patientExternalId} status=${status} ` +
        `patients=${totals.patientsUpserted} txns=${totals.transactionsWritten} ` +
        `dups=${totals.duplicates} facts(created=${totals.factsCreated} superseded=${totals.factsSuperseded})`,
    );

    return {
      runId,
      hostId: host.id,
      hostName: host.name,
      tenantIds: [...seenTenants].sort(),
      dryRun: false,
      syncLogId: syncLog.id,
      status,
      totals,
      perResource,
    };
  }

  // ─────────────────────────────────────────────────────────
  // Patients 主档(用同一份 yaml 形态,但走 upsert 路径)
  // ─────────────────────────────────────────────────────────

  private async processPatients(
    tables: Record<string, unknown[]>,
    config: AssemblerConfig,
    hostId: string,
    tenantResolver: TenantResolver,
    seenTenants: Set<string>,
    normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
    dryRun: boolean,
  ): Promise<PerResourceStats> {
    const stats: PerResourceStats = this.zeroPerResource(config.canonical);

    const assembled = this.assembler.assemble({ tables, config, normalize, hostId });
    stats.fetched = assembled.stats.fetched;
    stats.failed = assembled.stats.failed;
    if (assembled.rows.length > 0)
      stats.sampleCanonical.push(assembled.rows[0]!.canonical);

    for (const { canonical, rawSource } of assembled.rows) {
      const c = canonical as Record<string, unknown>;
      const externalId = String(c.externalId).trim();
      // v2.3:per-row tenant 解析(从 host 真原文 rawSource 读 brand 字段)
      const tenantId = tenantResolver.resolve(rawSource);
      if (!tenantId) {
        stats.failed++;
        this.logger.warn(
          `patient externalId=${externalId} 未匹配 tenant(${tenantResolver.describe()};rawSource 缺字段或值为空),跳过`,
        );
        continue;
      }
      seenTenants.add(tenantId);
      if (dryRun) {
        stats.patientsUpserted++;
        continue;
      }

      // 主表数据(身份)
      const patientData = {
        name: (c.name as string | undefined) ?? null,
        // ⚠️ DEMO 兜底:host 不给 phone(隐私) → 由 externalId hash 合成确定性假号
        //    (打 138 前缀,后 8 位是 externalId 数字 hash;同 externalId 多次跑得同号)
        //    生产 host 接入时把 c.phone 真值传过来即可,这里 ?? 自动让位真值
        phone: (c.phone as string | undefined) ?? synthesizeDemoPhone(externalId),
        gender: (c.gender as string | undefined) ?? null,
        birthDate: c.birthDate ? new Date(c.birthDate as string) : null,
        medicalRecordNumber:
          (c.medicalRecordNumber as string | undefined) ?? null,
        preferences:
          c.preferences && typeof c.preferences === 'object'
            ? (c.preferences as Prisma.InputJsonValue)
            : undefined,
        // status='archived'/'merged' → active=false;'active' / 缺省 → true
        active: ((c.status as string | undefined) ?? 'active') === 'active',
      };

      // 副表数据(合规/标签/状态)— 全部可空
      const profileData = {
        doNotContact: (c.doNotContact as boolean) ?? false,
        deceased: (c.deceased as boolean) ?? false,
        // canonical 字段(若 host 没给 → null/默认)
        // tags / notes / primaryContactType 需要从 canonical 拿;v2 canonical PatientSchema 暂不显式带
        // host 通过 host_tags / host_notes 等字段 → assembler.yaml 映射到 canonical;此处兜底
        tags: Array.isArray(c.tags) ? (c.tags as string[]) : [],
        notes: (c.notes as string | undefined) ?? null,
        primaryContactType:
          (c.primaryContactType as string | undefined) ?? null,
      };

      try {
        const patient = await this.prisma.patient.upsert({
          where: {
            hostId_tenantId_externalId: { hostId, tenantId, externalId },
          },
          create: { hostId, tenantId, externalId, ...patientData },
          update: patientData,
        });
        // 副表 1:1 upsert
        await this.prisma.patientProfile.upsert({
          where: { patientId: patient.id },
          create: { patientId: patient.id, ...profileData },
          update: profileData,
        });
        stats.patientsUpserted++;
      } catch (err) {
        stats.failed++;
        this.logger.error(
          `patient upsert failed externalId=${externalId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }

    return stats;
  }

  // ─────────────────────────────────────────────────────────
  // Subject 资源 → 合成 transaction → parser 衍生 fact
  // ─────────────────────────────────────────────────────────

  private async processSubject(
    tables: Record<string, unknown[]>,
    config: AssemblerConfig,
    hostId: string,
    tenantResolver: TenantResolver,
    seenTenants: Set<string>,
    normalize: { amountUnit: 'fen' | 'yuan'; timezone: string },
    syncLogId: string | undefined,
    dryRun: boolean,
  ): Promise<PerResourceStats> {
    const stats: PerResourceStats = this.zeroPerResource(config.canonical);

    const assembled = this.assembler.assemble({ tables, config, normalize, hostId });
    stats.fetched = assembled.stats.fetched;
    stats.failed += assembled.stats.failed; // assembler 自己的失败计在内
    if (assembled.rows.length > 0)
      stats.sampleCanonical.push(assembled.rows[0]!.canonical);

    // v2.3:per-tenant lazy patient index(多 tenant 时按需加载,避免一次拉全 host)
    const patientIndexByTenant = new Map<string, Map<string, string>>();
    // ⚠️ sourceContext 必须跨 run 稳定(否则 source_event_id 永远不重复)
    const sourceContext = `cold-import`;

    for (const { canonical, rawSource } of assembled.rows) {
      const canonicalRow = canonical as Record<string, unknown>;
      // per-row tenant 解析
      const tenantId = tenantResolver.resolve(rawSource);
      if (!tenantId) {
        stats.failed++;
        this.logger.warn(
          `${config.canonical} row 未匹配 tenant(${tenantResolver.describe()}),跳过`,
        );
        continue;
      }
      seenTenants.add(tenantId);
      if (dryRun) {
        stats.transactionsWritten++;
        continue;
      }
      // 懒加载该 tenant 的 patient index
      let patientIndex = patientIndexByTenant.get(tenantId);
      if (!patientIndex) {
        patientIndex = await this.buildPatientIndex(hostId, tenantId);
        patientIndexByTenant.set(tenantId, patientIndex);
      }
      // ⭐ W4 末 stub auto-create:本批 fact 引用的 patient 主档没拉到 → 即时建空 stub
      //   后续 fact_client_out(同 run 反向拉 OR 下次 sync 主档更新)会 upsert 补 PII
      //   保证 fact 立即派生不丢,不依赖等主档(详见 docs/dw-data-source-issues.md "patient stub")
      const patientExternalId = canonicalRow.patientExternalId as string | undefined;
      if (patientExternalId && !patientIndex.has(patientExternalId)) {
        const stubId = await this.ensurePatientStub(hostId, tenantId, patientExternalId);
        patientIndex.set(patientExternalId, stubId);
      }
      const txn = this.synthesizer.synthesize({
        rawRow: rawSource,
        canonicalRow,
        emits: assembled.emits,
        resource: config.canonical,
        hostId,
        tenantId,
        patientIdResolver: (extId) => patientIndex!.get(extId),
        sourceContext,
      });
      if (!txn) {
        stats.failed++;
        continue;
      }
      if (syncLogId) txn.syncLogId = syncLogId;

      try {
        const created = await this.prisma.patientTransaction.create({ data: txn });
        stats.transactionsWritten++;
        const metrics = await this.parserPipeline.runForTransaction({
          transaction: {
            id: created.id,
            hostId: created.hostId,
            tenantId: created.tenantId,
            patientId: created.patientId,
            action: created.action as Action,
            subjectType: created.subjectType,
            subjectId: created.subjectId,
            occurredAt: created.occurredAt,
            clinicId: created.clinicId,
          },
          canonicalRow,
        });
        stats.factsCreated += metrics.factsCreated;
        stats.factsSuperseded += metrics.factsSuperseded;
        stats.factsUnchanged += metrics.factsUnchanged;
        stats.factsEvidenceAppended += metrics.factsEvidenceAppended;
        stats.factsFailed += metrics.factsFailed;
      } catch (err) {
        const code = (err as { code?: string })?.code;
        if (code === 'P2002') {
          stats.duplicates++;
          // 幂等命中 — 仍跑 parser(中断恢复:transaction 落了但 fact 没落场景)
          const existing = await this.prisma.patientTransaction.findFirst({
            where: {
              hostId: txn.hostId,
              tenantId: txn.tenantId,
              sourceEventId: txn.sourceEventId ?? undefined,
            },
          });
          if (existing) {
            const metrics = await this.parserPipeline.runForTransaction({
              transaction: {
                id: existing.id,
                hostId: existing.hostId,
                tenantId: existing.tenantId,
                patientId: existing.patientId,
                action: existing.action as Action,
                subjectType: existing.subjectType,
                subjectId: existing.subjectId,
                occurredAt: existing.occurredAt,
                clinicId: existing.clinicId,
              },
              canonicalRow,
            });
            stats.factsCreated += metrics.factsCreated;
            stats.factsSuperseded += metrics.factsSuperseded;
            stats.factsUnchanged += metrics.factsUnchanged;
            stats.factsEvidenceAppended += metrics.factsEvidenceAppended;
            stats.factsFailed += metrics.factsFailed;
          }
        } else {
          stats.failed++;
          this.logger.error(
            `transaction write failed (resource=${config.canonical}, subject=${txn.subjectId}): ${
              err instanceof Error ? err.message : String(err)
            }`,
          );
        }
      }
    }

    return stats;
  }

  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 },
    });
    return new Map(rows.map((p) => [p.externalId, p.id]));
  }

  /// W4 末:本批 fact 引用的 patient 主档没拉到 → 即时建空 stub
  /// 只填三段隔离键(hostId/tenantId/externalId)+ active=true,姓名/电话留空待后续 upsert
  /// patientProfile 也建空行(scenario SQL JOIN 必须的 doNotContact=false / deceased=false)
  /// 调一次约 5-10 ms(单行 upsert + 1:1 profile upsert),N 次循环可接受;后续可批量优化
  private async ensurePatientStub(
    hostId: string,
    tenantId: string,
    externalId: string,
  ): Promise<string> {
    const patient = await this.prisma.patient.upsert({
      where: { hostId_tenantId_externalId: { hostId, tenantId, externalId } },
      create: { hostId, tenantId, externalId, active: true },
      update: {}, // 有则 noop(姓名等真实主档 upsert 走 processPatients 路径)
    });
    await this.prisma.patientProfile.upsert({
      where: { patientId: patient.id },
      create: { patientId: patient.id, doNotContact: false, deceased: false, tags: [] },
      update: {},
    });
    return patient.id;
  }

  // ─────────────────────────────────────────────────────────
  // Manifest + assembler + raw tables 读盘
  // ─────────────────────────────────────────────────────────

  private readManifest(dir: string): ColdImportManifest {
    const p = path.join(dir, 'manifest.yaml');
    if (!fs.existsSync(p)) throw new Error(`manifest.yaml not found at ${p}`);
    const parsed = yaml.load(fs.readFileSync(p, 'utf-8'));
    const result = ColdImportManifestSchema.safeParse(parsed);
    if (!result.success) {
      throw new Error(
        `manifest.yaml validation failed:\n${JSON.stringify(result.error.issues, null, 2)}`,
      );
    }
    return result.data;
  }

  /// W4 末:读最近一次成功的 incremental sync_log 的 cursor_after JSON
  /// 返回 { tableName: cursorValue } map(无记录返回空对象 → 首跑等价全量)
  private async readLastIncrementalCursor(hostId: string): Promise<Record<string, string>> {
    const last = await this.prisma.syncLog.findFirst({
      where: {
        hostId,
        resource: 'incremental_bundle',
        status: SyncStatus.SUCCESS,
        cursorAfter: { not: null },
      },
      orderBy: { startedAt: 'desc' },
    });
    if (!last?.cursorAfter) return {};
    try {
      const parsed = JSON.parse(last.cursorAfter) as Record<string, string>;
      return parsed;
    } catch (e) {
      this.logger.warn(`Cursor JSON parse failed: ${(e as Error).message};当全量处理`);
      return {};
    }
  }

  private async loadAllTables(
    dir: string,
    manifest: ColdImportManifest,
    incremental?: IncrementalConfig,
  ): Promise<Record<string, unknown[]>> {
    // ── ClickHouse 直连模式 ──
    if (manifest.sql_source) {
      this.logger.log(
        `数据源:ClickHouse SQL 直连 — ${manifest.sql_source.connection.url} ` +
          `database=${manifest.sql_source.connection.database} ` +
          `queries=${Object.keys(manifest.sql_source.queries).length}` +
          (incremental ? ' [incremental mode]' : ''),
      );
      return this.chSource.loadAllTables(manifest.sql_source, incremental);
    }

    // ── 文件源模式 ──
    if (!manifest.tables || manifest.tables.length === 0) {
      throw new Error(
        'manifest 缺数据源:既无 sql_source 也无 tables[](schema refine 本该拦住,兜底报错)',
      );
    }
    const tables: Record<string, unknown[]> = {};
    for (const t of manifest.tables) {
      const filePath = path.join(dir, t.file);
      if (!fs.existsSync(filePath)) {
        throw new Error(`Raw table file not found: ${filePath}`);
      }
      const format = t.format ?? inferFileFormat(t.file);
      tables[t.table] = this.readRows(filePath, format);
      this.logger.log(`  loaded raw table "${t.table}" (${tables[t.table]!.length} rows)`);
    }
    return tables;
  }

  private loadAllAssemblers(
    dir: string,
    manifest: ColdImportManifest,
  ): AssemblerConfig[] {
    const configs: AssemblerConfig[] = [];
    for (const ref of manifest.assemblers) {
      const filePath = path.join(dir, ref.file);
      if (!fs.existsSync(filePath)) {
        throw new Error(`Assembler config not found: ${filePath}`);
      }
      const parsed = yaml.load(fs.readFileSync(filePath, 'utf-8'));
      const r = AssemblerConfigSchema.safeParse(parsed);
      if (!r.success) {
        throw new Error(
          `Assembler ${ref.file} validation failed:\n${JSON.stringify(r.error.issues, null, 2)}`,
        );
      }
      configs.push(r.data);
      this.logger.log(`  loaded assembler "${r.data.canonical}" from ${ref.file}`);
    }
    return configs;
  }

  private readRows(filePath: string, format: 'csv' | 'json'): unknown[] {
    const raw = fs.readFileSync(filePath, 'utf-8');
    if (format === 'json') {
      const parsed = JSON.parse(raw);
      if (Array.isArray(parsed)) return parsed;
      if (Array.isArray(parsed?.items)) return parsed.items;
      throw new Error(
        `JSON file must be array or { items: [...] };got ${typeof parsed}`,
      );
    }
    return parseCsv(raw, {
      columns: true,
      skip_empty_lines: true,
      trim: true,
    });
  }

  // ─────────────────────────────────────────────────────────
  // Helpers
  // ─────────────────────────────────────────────────────────

  private zeroTotals(): TotalsBlock {
    return {
      patientsUpserted: 0,
      transactionsWritten: 0,
      duplicates: 0,
      failed: 0,
      factsCreated: 0,
      factsSuperseded: 0,
      factsUnchanged: 0,
      factsEvidenceAppended: 0,
      factsFailed: 0,
    };
  }

  private zeroPerResource(resource: string): PerResourceStats {
    return {
      resource,
      fetched: 0,
      patientsUpserted: 0,
      transactionsWritten: 0,
      duplicates: 0,
      failed: 0,
      factsCreated: 0,
      factsSuperseded: 0,
      factsUnchanged: 0,
      factsEvidenceAppended: 0,
      factsFailed: 0,
      sampleCanonical: [],
    };
  }
}

// ─────────────────────────────────────────────────────────
// Types
// ─────────────────────────────────────────────────────────

interface TotalsBlock {
  patientsUpserted: number;
  transactionsWritten: number;
  duplicates: number;
  failed: number;
  factsCreated: number;
  factsSuperseded: number;
  factsUnchanged: number;
  factsEvidenceAppended: number;
  factsFailed: number;
}

export interface PerResourceStats extends TotalsBlock {
  resource: string;
  fetched: number;
  sampleCanonical: unknown[];
}

/**
 * 给 externalId 算确定性"假手机号"(demo 兜底,host 真给的 phone 走真值)。
 * 算法:138 + (externalId 转 number) mod 1e8,补 0 至 8 位 → 138 + 8 位 = 11 位
 * 同 externalId 多次 import 出同号(便于 demo / debug)。
 */
function synthesizeDemoPhone(externalId: string): string {
  let hash = 0;
  for (let i = 0; i < externalId.length; i++) {
    hash = (hash * 31 + externalId.charCodeAt(i)) >>> 0;
  }
  const last8 = String(hash % 100_000_000).padStart(8, '0');
  return `138${last8}`;
}

export interface ImportRunResult {
  runId: string;
  hostId: string;
  hostName: string;
  /// v2.3:多 tenant per host 时这里是所有 known tenant 列表;单 tenant 时长度为 1
  tenantIds: string[];
  dryRun: boolean;
  syncLogId?: string;
  status: string;
  totals: TotalsBlock;
  perResource: PerResourceStats[];
}
