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, type BatchItem } from '../pipeline/parser-pipeline.service';
import {
  AssemblerEngine,
  type AssemblerResult,
  type MappingMiss,
  type SuspectField,
  mergeMappingMisses,
  mergeSuspectFields,
} from '../assembler/assembler-engine';
import {
  AssemblerConfigSchema,
  type AssemblerConfig,
} from '../assembler/assembler.schema';
import {
  ColdImportManifest,
  ColdImportManifestSchema,
  inferFileFormat,
} from './manifest.schema';
import {
  ClickHouseSourceService,
  type CohortKey,
  type IncrementalConfig,
  type PatientScope,
} from './clickhouse-source.service';
import { AlertService } from '../../../common/alerting/alert.service';
import { TransformEngine } from '../transforms/transform-engine';
import { buildTenantResolver, type TenantResolver } from './tenant-resolver';
import { mergePatientPreferences } from '../pipeline/patient-preferences.util';

/**
 * 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,
    private readonly alerter: AlertService,
  ) {}

  /**
   * Reparse — 用已存的 `transaction.rawPayload` 离线重跑 装配+解析(**不连 DW**)。
   *
   * 场景:改了 assembler yaml(enum_mapping / keyword_mapping / field_mapping / transforms)后,
   * 把存量 patient_facts 按新口径重新衍生 —— **不 truncate、不碰 transaction 账本、不拉 DW**,
   * 因此执行历史(plan_executions)/ 分配全保住,比全量重摄快一个量级(省掉 DW 拉取)。
   *
   * 原理:`transaction.rawPayload` 存的就是 host 原始源行(如 EMR 行)。重建出原始源表 →
   * 喂回 `transformEngine` + `processSubject`;transaction 经 source_event_id 幂等命中已存
   * (createMany skipDuplicates 跳过重写),但 SELECT-back 后 **parser 照常重跑** → fact 走
   * 版本流(内容变 → supersede 升版本,不变 → no-op)。
   *
   * 覆盖:field/enum/keyword_mapping + transforms 算子 + parser 改动(长尾字典修补 99%)。
   * 不覆盖:改了 CH `sql_source` SELECT 本身(要从 DW 多拉原来没拉的列/表)→ 那才需真重摄。
   *
   * @returns 受影响 patientId(调用方据此重算 persona/plan);dryRun 时只 diff 不写。
   */
  async reparseFromTransactions(opts: {
    dir: string;
    hostName: string;
    subjectTypes?: string[]; // 限定 canonical(如 ['diagnosis']);空 = 全部 subject 资源
    patientIds?: string[];
    dryRun?: boolean;
  }): Promise<{
    perResource: PerResourceStats[];
    affectedPatientIds: string[];
    dryRunDiffs: Array<{ resource: string; subjectId: string; field: string; from: string; to: string }>;
  }> {
    const absDir = path.resolve(opts.dir);
    const manifest = this.readManifest(absDir);
    const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
    if (!host) throw new Error(`reparse: host '${opts.hostName}' not found`);
    const tenantResolver = buildTenantResolver(manifest);
    const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
    const subjectCfgs = this.loadAllAssemblers(absDir, manifest).filter((c) => !!c.emits);
    const targetCfgs = opts.subjectTypes?.length
      ? subjectCfgs.filter((c) => opts.subjectTypes!.includes(c.canonical))
      : subjectCfgs;
    if (targetCfgs.length === 0) {
      throw new Error(
        `reparse: 无匹配 subject 配置(subjectTypes=${opts.subjectTypes?.join(',') ?? 'all'});` +
          `可选=${subjectCfgs.map((c) => c.canonical).join(',')}`,
      );
    }
    const transforms = manifest.transforms ?? [];
    // 只能离线 reparse「transform 产出的 primary 表」:其 rawPayload 是 transform 的源行,可重放。
    // 非 transform 产出的(如 image_finding_rows = CH SQL 视图)→ rawPayload 不是其装配输入 → 跳过(需走 DW)。
    const reparseableCfgs = targetCfgs.filter((cfg) => {
      const raw = traceRawSourceTable(cfg.primary.table, transforms);
      if (raw === cfg.primary.table) {
        this.logger.warn(
          `reparse: 跳过 ${cfg.canonical}(file primary=${cfg.primary.table} 非 transform 产出 → ` +
            `rawPayload 不是其装配输入,无法离线 reparse;该资源如需更新走 DW 重摄)`,
        );
        return false;
      }
      return true;
    });
    if (reparseableCfgs.length === 0) {
      throw new Error('reparse: 目标资源都非 transform 产出,无法离线 reparse(需走 DW 重摄)');
    }
    this.logger.log(
      `reparse: host=${host.name} resources=[${reparseableCfgs.map((c) => c.canonical + '/' + c.primary.table).join(',')}] ` +
        `dryRun=${!!opts.dryRun} patientScope=${opts.patientIds?.length ?? 'all'}`,
    );

    // 1. 解析 patient 范围 —— **按患者分批**重建源表,避免一次性把全 host rawPayload(可达 GB 级)
    //    拉进内存 OOM(全量 reparse 时尤为关键)。
    const subjectTypes = reparseableCfgs.map((c) => c.emits!.subjectType);
    let scopePatientIds = opts.patientIds;
    if (!scopePatientIds?.length) {
      const rows = await this.prisma.patientTransaction.findMany({
        where: { hostId: host.id, subjectType: { in: subjectTypes }, patientId: { not: null } },
        select: { patientId: true },
        distinct: ['patientId'],
      });
      scopePatientIds = rows.map((r) => r.patientId).filter((x): x is string => !!x);
    }
    const dryRunDiffs: Array<{ resource: string; subjectId: string; field: string; from: string; to: string }> = [];

    // 2a. dryRun:报 scope(患者数 + 各资源 txn 数),不写、不分批装配(便宜)。
    if (opts.dryRun) {
      for (const cfg of reparseableCfgs) {
        const n = await this.prisma.patientTransaction.count({
          where: { hostId: host.id, subjectType: cfg.emits!.subjectType, ...(opts.patientIds?.length ? { patientId: { in: opts.patientIds } } : {}) },
        });
        this.logger.log(`reparse[dry]: ${cfg.canonical}(${cfg.emits!.subjectType}) txns=${n} → 实跑按版本流 supersede 变更的、跳过不变的`);
      }
      this.logger.log(`reparse[dry]: 范围 ${scopePatientIds.length} 患者;去掉 --dry-run 实跑(非破坏)。`);
      return { perResource: [], affectedPatientIds: [], dryRunDiffs };
    }

    // 2b. 分批实跑:每批患者 → 重建源表(只这批 distinct rawPayload)→ transform → processSubject
    //     (transaction 幂等命中已存 → parser 重衍生 fact,版本流 supersede)。
    const BATCH = Math.max(1, Number(process.env.PAC_REPARSE_BATCH) || 3000);
    const runStart = new Date();
    const aggByResource = new Map<string, PerResourceStats>();
    const seenTenants = new Set<string>();
    const totalBatches = Math.ceil(scopePatientIds.length / BATCH);
    this.logger.log(`reparse: 范围 ${scopePatientIds.length} 患者,分 ${totalBatches} 批(${BATCH}/批)`);
    for (let bi = 0; bi < totalBatches; bi++) {
      const batch = scopePatientIds.slice(bi * BATCH, (bi + 1) * BATCH);
      const tables: Record<string, Record<string, unknown>[]> = {};
      const seenHashByTable = new Map<string, Set<string>>();
      for (const cfg of reparseableCfgs) {
        const rawTable = traceRawSourceTable(cfg.primary.table, transforms);
        const bucket = (tables[rawTable] ??= []);
        const seen = seenHashByTable.get(rawTable) ?? new Set<string>();
        seenHashByTable.set(rawTable, seen);
        const txs = await this.prisma.patientTransaction.findMany({
          where: { hostId: host.id, subjectType: cfg.emits!.subjectType, patientId: { in: batch } },
          select: { rawPayload: true, payloadHash: true },
        });
        for (const t of txs) {
          if (seen.has(t.payloadHash)) continue;
          seen.add(t.payloadHash);
          bucket.push(t.rawPayload as Record<string, unknown>);
        }
      }
      for (const tf of transforms) {
        const inp = (tf as { input?: string }).input;
        if (inp && !tables[inp]) tables[inp] = [];
      }
      const transformed = transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;
      for (const cfg of reparseableCfgs) {
        const stats = await this.processSubject(transformed, cfg, host.id, tenantResolver, seenTenants, normalize, undefined, false);
        const agg = aggByResource.get(cfg.canonical);
        if (!agg) {
          aggByResource.set(cfg.canonical, stats);
        } else {
          agg.transactionsWritten += stats.transactionsWritten;
          agg.duplicates += stats.duplicates;
          agg.failed += stats.failed;
          agg.factsCreated += stats.factsCreated;
          agg.factsSuperseded += stats.factsSuperseded;
          agg.factsUnchanged += stats.factsUnchanged;
          agg.factsEvidenceAppended += stats.factsEvidenceAppended;
          agg.factsFailed += stats.factsFailed;
          agg.fetched += stats.fetched;
          agg.mappingMisses = mergeMappingMisses([agg.mappingMisses, stats.mappingMisses]);
          agg.suspectFields = mergeSuspectFields([agg.suspectFields, stats.suspectFields]);
        }
      }
      if ((bi + 1) % 5 === 0 || bi + 1 === totalBatches) {
        const sup = [...aggByResource.values()].reduce((s, r) => s + r.factsSuperseded, 0);
        this.logger.log(
          `reparse: batch ${bi + 1}/${totalBatches} 累计 superseded=${sup} ` +
            `rss=${Math.round(process.memoryUsage().rss / 1024 / 1024)}MB`,
        );
      }
    }
    const perResource = [...aggByResource.values()];

    // 保障机制:摄入漂移审计 —— 全量 reparse 是看"还有哪些原值没映射上 / 哪些列消失了"的最佳时机。
    const allMisses = mergeMappingMisses(perResource.map((s) => s.mappingMisses));
    if (allMisses.length > 0) {
      this.logger.warn(
        `reparse 映射覆盖漏 ${allMisses.length} 种(落 _default,按量倒序 top 20):\n` +
          allMisses
            .slice(0, 20)
            .map((m) => `  ${m.field}='${m.rawValue}' ×${m.count} → '${m.fellBackTo}'`)
            .join('\n'),
      );
    }
    const allSuspect = mergeSuspectFields(perResource.map((s) => s.suspectFields));
    if (allSuspect.length > 0) {
      this.logger.warn(
        `⚠️ reparse 疑似宿主改表 ${allSuspect.length} 列整列缺席:\n` +
          allSuspect.map((s) => `  ${s.resource}.${s.field}(host 列 ${s.hostField})`).join('\n'),
      );
    }

    // 3. 受影响 patientId = 本次真正被 supersede(内容变了)的 fact 的 distinct patient → 只重算这些。
    const changed = await this.prisma.patientFact.findMany({
      where: {
        hostId: host.id,
        supersededAt: { gte: runStart },
        ...(opts.patientIds?.length ? { patientId: { in: opts.patientIds } } : {}),
      },
      select: { patientId: true },
      distinct: ['patientId'],
    });
    const affectedPatientIds = changed.map((a) => a.patientId).filter((x): x is string => !!x);
    return { perResource, affectedPatientIds, dryRunDiffs };
  }

  /**
   * 形态 A push 入口:宿主推来某张【原生源表】的原生行 → 复用 transforms + assembler + processSubject 落库。
   *
   * 跟 cold-import / reparse 同一个核(raw → canonical → txn → fact 版本流),唯一差异:
   *   原生行来自 webhook 请求,而不是 ClickHouse SQL / transactions.rawPayload。
   *
   * 不触发 persona / plan(由 caller 决定);返回 touched 患者(带 tenant)供 caller enqueue 画像重算。
   * tenant 走 manifest 的 brand resolver(每行解析),与 pull / reparse 完全一致。
   * 幂等:processSubject 内 sourceContext 固定,source_event_id 与 pull 同口径去重。
   */
  async ingestRawTables(opts: {
    hostName: string;
    source: string; // 原生源表名(= manifest tables[].table / transform 终端表)
    rows: Record<string, unknown>[];
    triggeredBy?: string;
  }): Promise<{
    syncLogId: string;
    perResource: PerResourceStats[];
    touched: Array<{ patientId: string; tenantId: string }>;
  }> {
    const absDir = this.resolveHostDir(opts.hostName);
    const manifest = this.readManifest(absDir);
    const host = await this.prisma.host.findFirst({ where: { name: opts.hostName } });
    if (!host) throw new Error(`ingestRawTables: host '${opts.hostName}' not found`);
    const tenantResolver = buildTenantResolver(manifest);
    const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };
    const transforms = manifest.transforms ?? [];

    // 找出"终端原生表 === source"的所有 subject cfg(一张 EMR 源表可喂 诊断/治疗/emr/建议 多 cfg)
    const subjectCfgs = this.loadAllAssemblers(absDir, manifest).filter((c) => !!c.emits);
    const targetCfgs = subjectCfgs.filter(
      (cfg) => traceRawSourceTable(cfg.primary.table, transforms) === opts.source,
    );
    if (targetCfgs.length === 0) {
      const known = [
        ...new Set(subjectCfgs.map((c) => traceRawSourceTable(c.primary.table, transforms))),
      ];
      throw new Error(
        `ingestRawTables: source='${opts.source}' 无匹配 assembler(可选源表=${known.join(',')})`,
      );
    }

    const syncLog = await this.prisma.syncLog.create({
      data: {
        hostId: host.id,
        tenantId: 't_pending', // push 多 tenant 由行决定;finalize 时回填首个 seen tenant
        direction: SyncDirection.PUSH,
        resource: opts.source,
        triggeredBy: opts.triggeredBy ?? `push:${opts.hostName}:${opts.source}`,
        status: SyncStatus.RUNNING,
      },
    });

    // ⚠️ transform 可能 in-place 改源行(剥未知列),故在 transform【之前】快照推送列集,供 raw 层列漂移检测。
    const pushedCols = new Set<string>();
    for (const r of opts.rows) for (const k of Object.keys(r)) if (!k.startsWith('_')) pushedCols.add(k);

    // tables = { [source]: rows } + transform 空输入兜底 → transform → processSubject
    const tables: Record<string, Record<string, unknown>[]> = { [opts.source]: opts.rows };
    for (const tf of transforms) {
      const inp = (tf as { input?: string }).input;
      if (inp && !tables[inp]) tables[inp] = [];
    }
    const transformed =
      transforms.length > 0 ? this.transformEngine.run({ tables, transforms }) : tables;

    const seenTenants = new Set<string>();
    const perResource: PerResourceStats[] = [];
    let firstError: string | null = null;
    try {
      for (const cfg of targetCfgs) {
        const stats = await this.processSubject(
          transformed,
          cfg,
          host.id,
          tenantResolver,
          seenTenants,
          normalize,
          syncLog.id,
          false,
        );
        perResource.push(stats);
      }
    } catch (err) {
      firstError = err instanceof Error ? err.message : String(err);
      this.logger.error(`ingestRawTables(${opts.source}) 异常: ${firstError}`);
    }

    // touched 患者 = 本 syncLog 写入的 transaction 的 distinct (patientId, tenantId)
    const touchedRows = await this.prisma.patientTransaction.findMany({
      where: { syncLogId: syncLog.id, patientId: { not: null } },
      select: { patientId: true, tenantId: true },
      distinct: ['patientId'],
    });
    const touched = touchedRows
      .filter((t): t is { patientId: string; tenantId: string } => !!t.patientId)
      .map((t) => ({ patientId: t.patientId, tenantId: t.tenantId }));

    const agg = perResource.reduce(
      (a, s) => ({
        txn: a.txn + s.transactionsWritten,
        dup: a.dup + s.duplicates,
        failed: a.failed + s.failed,
        facts: a.facts + s.factsCreated + s.factsSuperseded,
      }),
      { txn: 0, dup: 0, failed: 0, facts: 0 },
    );
    // ── raw 层列漂移检测(form A 专属,最可靠):本次推送行的列集 vs 历史同源 rawPayload 列集 ──
    //   transform 会掩盖 assembler 层的删列(补 key),故在原生行层面直接对比。基线 = 已存 transaction,
    //   不需新存储。基线样本太少(新 host 冷启)则跳过,避免误报。
    const rawDrift = await this.detectRawColumnDrift({
      hostId: host.id,
      source: opts.source,
      subjectTypes: targetCfgs.map((c) => c.emits!.subjectType),
      pushedCols,
      rowCount: opts.rows.length,
      excludeSyncLogId: syncLog.id, // ⚠️ 基线排除本 run 刚写的行(否则加列会被自己污染成基线)
    });

    // 摄入漂移(保障机制)→ SyncLog.metadata,供审计:
    //   mappingMisses = 落 _default 的原值(扩字典);suspectFields = 疑似宿主改表(列缺席/删/加)。
    const mappingMisses = mergeMappingMisses(perResource.map((s) => s.mappingMisses));
    const suspectFields = mergeSuspectFields([
      ...perResource.map((s) => s.suspectFields),
      rawDrift,
    ]);
    if (mappingMisses.length > 0) {
      this.logger.warn(
        `ingestRawTables(${opts.source}) 映射覆盖漏 ${mappingMisses.length} 种:` +
          mappingMisses
            .slice(0, 5)
            .map((m) => `${m.field}='${m.rawValue}'×${m.count}`)
            .join(', ') +
          (mappingMisses.length > 5 ? ' …' : ''),
      );
    }
    if (suspectFields.length > 0) {
      this.logger.warn(
        `⚠️ ingestRawTables(${opts.source}) 疑似宿主改表 ${suspectFields.length} 项:` +
          suspectFields.map((s) => `[${s.reason}] ${s.resource}.${s.field}`).join(', '),
      );
    }
    const driftMeta =
      mappingMisses.length > 0 || suspectFields.length > 0
        ? {
            ...(mappingMisses.length > 0 ? { mappingMisses } : {}),
            ...(suspectFields.length > 0 ? { suspectFields } : {}),
          }
        : null;

    await this.prisma.syncLog.update({
      where: { id: syncLog.id },
      data: {
        status:
          firstError && agg.txn === 0
            ? SyncStatus.FAILED
            : firstError
              ? SyncStatus.PARTIAL
              : SyncStatus.SUCCESS,
        tenantId: [...seenTenants][0] ?? 't_unknown',
        fetched: opts.rows.length,
        transactionsWritten: agg.txn,
        factsEmitted: agg.facts,
        duplicates: agg.dup,
        failed: agg.failed,
        errorMessage: firstError,
        endedAt: new Date(),
        ...(driftMeta ? { metadata: driftMeta as unknown as Prisma.InputJsonValue } : {}),
      },
    });

    this.logger.log(
      `ingestRawTables host=${opts.hostName} source=${opts.source} rows=${opts.rows.length} ` +
        `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} touched=${touched.length}`,
    );
    return { syncLogId: syncLog.id, perResource, touched };
  }

  /// host → 配置目录(= sync.service.resolveDataDir 同约定:env PAC_INCREMENTAL_DATA_DIR ?? cwd/data,再拼 hostName)
  private resolveHostDir(hostName: string): string {
    const base = process.env.PAC_INCREMENTAL_DATA_DIR ?? path.resolve(process.cwd(), 'data');
    return path.join(base, hostName);
  }

  /**
   * raw 层列漂移检测(form A):本次推送行的列集 vs 历史同源 rawPayload 的列集。
   *   - 历史有、本次没有 → column_removed(疑似宿主删列/改名)
   *   - 本次有、历史没有 → column_added(疑似宿主加列/改名)
   * 基线 = 已存 transaction 的 rawPayload(同 subjectType 抽样 50 条),不需新存储。
   * 基线样本 < 5 条(新 host 冷启)→ 跳过,避免误报。`_` 开头的内部 meta key 忽略。
   */
  private async detectRawColumnDrift(args: {
    hostId: string;
    source: string;
    subjectTypes: string[];
    pushedCols: Set<string>;
    rowCount: number;
    excludeSyncLogId: string;
  }): Promise<SuspectField[]> {
    const MIN_BASELINE = 5;
    const sample = await this.prisma.patientTransaction.findMany({
      where: {
        hostId: args.hostId,
        subjectType: { in: args.subjectTypes },
        syncLogId: { not: args.excludeSyncLogId }, // 排除本 run 刚写的行(防自污染)
      },
      select: { rawPayload: true },
      take: 50,
      orderBy: { createdAt: 'desc' },
    });
    if (sample.length < MIN_BASELINE) return [];
    const baseline = new Set<string>();
    for (const s of sample) {
      const rp = s.rawPayload as Record<string, unknown> | null;
      if (rp) for (const k of Object.keys(rp)) if (!k.startsWith('_')) baseline.add(k);
    }

    const out: SuspectField[] = [];
    for (const c of baseline)
      if (!args.pushedCols.has(c))
        out.push({ resource: args.source, field: c, hostField: c, total: args.rowCount, reason: 'column_removed' });
    for (const c of args.pushedCols)
      if (!baseline.has(c))
        out.push({ resource: args.source, field: c, hostField: c, total: args.rowCount, reason: 'column_added' });
    return out;
  }

  async importDirectory(
    dir: string,
    options: {
      dryRun?: boolean;
      incremental?: boolean;
      /// 按 patient 分批跑(内存友好)。 null/undefined/0 → 单 shot(向后兼容);
      /// 实际生效大小见 resolveCohortBatchSize(env PAC_COHORT_BATCH_SIZE 兜底,默认 5000)
      cohortBatchSize?: number | null;
    } = {},
  ): Promise<ImportRunResult> {
    const absDir = path.resolve(dir);
    const runId = randomUUID();
    // ⭐ run_start 在函数入口冻结,作为 cursor_after baseline(见下方 §finalize)
    //   保证存量跑期间任何 DW 新写入(无论已摄入 / 未摄入患者),都能被下次增量 cron 捞回
    const runStart = new Date();
    this.logger.log(
      `Sync starting: ${absDir} (dryRun=${!!options.dryRun}, ` +
        `incremental=${!!options.incremental}, runId=${runId}, run_start=${runStart.toISOString()})`,
    );

    // 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. 增量 cursor 准备(PR3 统一:只要 manifest 有 incremental.per_query 配置,始终读 + 写 cursor)
    //    options.incremental 兼容老调用方:
    //      true(默认)→ 读上次 cursor,SQL WHERE > cursor 增量
    //      false      → 忽略上次 cursor(等价 ignoreCursor=true),但仍写新 cursor_after=run_start
    //                   用于灾后重灌 / 强制全量 / cursor 损坏修复
    //    file 源(无 sql_source)→ 永远 single-shot,不参与 cursor
    let incrementalConfig: IncrementalConfig | undefined;
    let cursorBeforeJson: string | null = null;
    const perQueryCfg = manifest.sql_source?.incremental?.per_query;
    if (perQueryCfg) {
      // 默认读 cursor;options.incremental === false 时 (强制 full) 不读
      const ignoreCursor = options.incremental === false;
      const lastCursor = ignoreCursor
        ? {} as Record<string, string>
        : await this.readLastIncrementalCursor(host.id);
      this.logger.log(
        `Cursor:${ignoreCursor ? '(显式忽略 — --full)' : Object.keys(lastCursor).length === 0 ? '(首跑,全量)' : JSON.stringify(lastCursor)}`,
      );
      incrementalConfig = {
        perQuery: Object.fromEntries(
          Object.entries(perQueryCfg).map(([table, cfg]) => [
            table,
            {
              cursorColumn: cfg.cursor_column,
              // ⚠️ 必须规范化:DW 的 cursor 列是 String(如 updated_date: Nullable(String),
              // 值形如 'YYYY-MM-DD HH:mm:ss')→ WHERE 比较是字典序。游标存的是 run_start ISO
              // ('2026-06-09T03:30:00.012Z'),第 11 字符 'T'(0x54) > ' '(0x20) → 所有真实行
              // 都"小于"游标 → 增量永远 fetched=0(2026-06-10 服务器实测三天空转)。
              // 这里读取侧转成 DW 时区的同款格式,旧游标无需迁移。
              cursorValue: lastCursor[table]
                ? toDwCursorLiteral(lastCursor[table], manifest.timezone)
                : null,
            },
          ]),
        ),
      };
      cursorBeforeJson = JSON.stringify(
        Object.fromEntries(
          Object.entries(incrementalConfig.perQuery).map(([k, v]) => [k, v.cursorValue ?? '']),
        ),
      );
    } else if (options.incremental) {
      // 老调用方明确要 incremental 但 manifest 没配 cursor → fast-fail
      throw new Error(
        `incremental 模式需要 manifest.sql_source.incremental.per_query;请补 cursor 配置`,
      );
    }

    // 4. SyncLog 创建 — ⭐ 作为并发锁
    //   sync_logs 表有 partial UNIQUE (host_id) WHERE status='running'(migration 20260528000000)
    //   同 host 已有 running 行时,本次 INSERT 撞 P2002 → 抛 SyncAlreadyRunningError → 调用方 skip
    //   finally 块统一 finalize status,确保锁一定被释放(进程崩需依赖 stale 清理)
    const syncLogTenant = knownTenants.length === 1 ? knownTenants[0]! : '_multi';
    // PR3 统一 resource = 'incremental_bundle';legacy --full 也写此 resource,
    // 这样 readLastIncrementalCursor 能从任一历史 sync_log 拿到 cursor,语义统一
    const syncResource = 'incremental_bundle';
    const triggerMode = options.incremental === false ? 'full' : 'sync';
    let syncLog: { id: string } | null = null;
    if (!options.dryRun) {
      try {
        syncLog = await this.prisma.syncLog.create({
          data: {
            hostId: host.id,
            tenantId: syncLogTenant,
            direction: SyncDirection.PULL, // 冷启复用 pull 语义(同套 pipeline)
            resource: syncResource,
            triggeredBy: `${triggerMode}:${path.basename(absDir)}:${runId}`,
            status: SyncStatus.RUNNING,
            cursorBefore: cursorBeforeJson,
          },
          select: { id: true },
        });
      } catch (err) {
        const code = (err as { code?: string }).code;
        if (code === 'P2002') {
          // 并发拦截 — 已有同 host 的 running sync_log,本次直接抛弃
          const existing = await this.prisma.syncLog.findFirst({
            where: { hostId: host.id, status: SyncStatus.RUNNING },
            orderBy: { startedAt: 'desc' },
            select: { id: true, startedAt: true, triggeredBy: true },
          });
          this.logger.warn(
            `Sync 已在跑(host=${host.name}) — 本次 skip(partial-unique 锁拦截)。` +
              `existing sync_log id=${existing?.id} started_at=${existing?.startedAt.toISOString()} ` +
              `triggered_by=${existing?.triggeredBy}`,
          );
          throw new SyncAlreadyRunningError(host.name, existing);
        }
        throw err;
      }
    }

    // ─── 进入 try/finally:任何抛错都要走 finally finalize syncLog,释放锁 ───
    const perResource: PerResourceStats[] = [];
    const totals = this.zeroTotals();
    let firstError: string | null = null;
    let status: string = SyncStatus.FAILED;

    try {
      // 5. 加载 assembler configs(一次,全 cohort batch 共用)
      const assemblerConfigs = this.loadAllAssemblers(absDir, manifest);
      const patientCfg = assemblerConfigs.find((c) => c.canonical === 'patient');
      const patientRelationCfg = assemblerConfigs.find((c) => c.canonical === 'patient_relation');
    const patientReturnVisitCfg = assemblerConfigs.find((c) => c.canonical === 'patient_return_visit');
      // patient + patient_relation 都是 upsert 资源(非 transaction)→ 排除出 subjectCfgs
      const subjectCfgs = assemblerConfigs.filter(
        (c) =>
        c.canonical !== 'patient' &&
        c.canonical !== 'patient_relation' &&
        c.canonical !== 'patient_return_visit',
      );
      const normalize = { amountUnit: manifest.amount_unit, timezone: manifest.timezone };

      // 6. 决定 cohort 模式:
      //   - cohortBatchSize > 0 + ClickHouse 源 + 配了 cohort → 分批跑(内存友好,resume 友好)
      //   - 未设 / 文件源 / 没配 cohort → 单 shot(向后兼容,csv 模式数据量小不需分批)
      const cohortBatchSize = resolveCohortBatchSize(options.cohortBatchSize);
      const canCohort = cohortBatchSize > 0 && !!manifest.sql_source?.cohort;
      if (cohortBatchSize > 0 && manifest.sql_source && !manifest.sql_source.cohort) {
        this.logger.warn(
          `cohortBatchSize=${cohortBatchSize} 但 manifest.sql_source.cohort 未配置 — ` +
            `退回 single-shot(要分批请在 manifest 补 cohort 段)`,
        );
      }
      if (canCohort && manifest.sql_source) {
        const sqlSource = manifest.sql_source;
        // ── cohort 模式 ──
        // 6a. 列出全部 patient pair(增量 cursor 同样适用)
        const allPairs = await this.chSource.listPatientPairs(
          sqlSource,
          incrementalConfig,
        );
        if (allPairs.length === 0) {
          this.logger.log(`Cohort: 0 patients in scope(增量空跑或源数据为空)— skip 全 batch`);
        } else {
          const batches = chunk(allPairs, cohortBatchSize);
          const concurrency = resolveCohortConcurrency();
          // ⭐ PR5b:并行模式不做 checkpoint resume(完成顺序乱,前缀语义失效)
          //   靠 source_event_id 幂等,崩溃从头重跑安全;串行模式保留 resume
          const resumeFrom = concurrency > 1 ? 0 : await this.readCheckpointOffset(syncLog?.id);
          this.logger.log(
            `Cohort 模式:patients=${allPairs.length} batchSize=${cohortBatchSize} ` +
              `batches=${batches.length} concurrency=${concurrency}` +
              (resumeFrom > 0 ? ` resumeFrom=${resumeFrom}` : '') +
              (concurrency > 1 ? ' (并行:checkpoint resume 禁用,靠幂等)' : ''),
          );

          let doneCount = resumeFrom;
          const runOneBatch = async (i: number): Promise<void> => {
            const batchStart = Date.now();
            const batch = batches[i]!;
            this.logger.log(
              `─── cohort batch ${i + 1}/${batches.length} start ` +
                `(${batch.length} patients) ─────`,
            );
            await this.processCohort({
              absDir,
              manifest,
              host,
              tenantResolver,
              seenTenants,
              normalize,
              patientCfg,
              patientRelationCfg,
              patientReturnVisitCfg,
              subjectCfgs,
              syncLogId: syncLog?.id,
              dryRun: options.dryRun ?? false,
              cohort: batch,
              incrementalConfig,
              perResource,
              totals,
              onFirstError: (msg) => { if (!firstError) firstError = msg; },
            });
            doneCount++;
            // checkpoint 写入(串行=精确前缀;并行=count-based 近似进度)
            if (syncLog) {
              await this.prisma.syncLog.update({
                where: { id: syncLog.id },
                data: {
                  metadata: {
                    cohortTotal: batches.length,
                    cohortDone: doneCount,
                    cohortMode: concurrency > 1 ? 'parallel' : 'serial',
                    concurrency,
                    batchSize: cohortBatchSize,
                    lastBatchMs: Date.now() - batchStart,
                    lastBatchRssMb: Math.round(process.memoryUsage().rss / 1024 / 1024),
                    lastBatchAt: new Date().toISOString(),
                  },
                },
              });
            }
            this.logger.log(
              `─── cohort batch ${i + 1}/${batches.length} done ${Date.now() - batchStart}ms ` +
                `rss=${Math.round(process.memoryUsage().rss / 1024 / 1024)}MB ` +
                `(${doneCount}/${batches.length}) ─────`,
            );
          };

          const indices = Array.from(
            { length: batches.length - resumeFrom },
            (_, k) => resumeFrom + k,
          );
          if (concurrency <= 1) {
            // 串行(默认,行为不变)
            for (const i of indices) await runOneBatch(i);
          } else {
            // 并行 worker pool — N 个 runner 各自从 indices 队列拉下一个直到耗尽
            let cursor = 0;
            const runner = async (): Promise<void> => {
              while (cursor < indices.length) {
                const myIdx = indices[cursor++]!;
                await runOneBatch(myIdx);
              }
            };
            await Promise.all(
              Array.from({ length: Math.min(concurrency, indices.length) }, () => runner()),
            );
          }
        }
      } else {
        // ── single-shot 模式(向后兼容 / 文件源)──
        await this.processCohort({
          absDir,
          manifest,
          host,
          tenantResolver,
          seenTenants,
          normalize,
          patientCfg,
          patientRelationCfg,
          patientReturnVisitCfg,
          subjectCfgs,
          syncLogId: syncLog?.id,
          dryRun: options.dryRun ?? false,
          cohort: null,
          incrementalConfig,
          perResource,
          totals,
          onFirstError: (msg) => { if (!firstError) firstError = msg; },
        });
      }

      status =
        totals.failed === 0
          ? SyncStatus.SUCCESS
          : totals.patientsUpserted + totals.transactionsWritten > 0
            ? SyncStatus.PARTIAL
            : SyncStatus.FAILED;
    } catch (workErr) {
      // 重大工作异常(载入失败 / transform 引擎崩 / DB 整体故障等)→ failed,但仍走 finally 释放锁
      const msg = workErr instanceof Error ? workErr.message : String(workErr);
      firstError ||= `fatal: ${msg}`;
      status = SyncStatus.FAILED;
      this.logger.error(`Sync work fatal: ${msg}`);
      // 不 rethrow — 让 finally 把状态写回 DB,然后再 throw 给调用方
      // 但调用方需要知道这次失败,所以 finally 之后重抛
      // eslint-disable-next-line no-unsafe-finally,@typescript-eslint/no-unsafe-finally
    } finally {
      // ⭐ 不管成功失败,把 syncLog 标 final status(释放 partial-unique 锁)
      // ⭐ cursor_after = run_start ISO(关键!不是 max(updated_date) — 避免漏批次间 DW 新增)
      //   理由:run_start 之后 DW 任何写入,下次增量 WHERE > run_start 都能捞回
      //   - 同行不同 updatedAt → source_event_id 不同 → 新 tx + 新 fact 版本(parser supersede 旧版)
      //   - 同行同 updatedAt → P2002 conflict → parser re-run(idempotent,无害)
      // PR3 统一:只要 manifest 配了 incremental.per_query,无论 mode(sync / --full)
      // 都写 cursor_after = run_start。--full 也写,保证下次正常 sync 从这里接力,
      // 不会因为 --full 没写 cursor 而漏掉 --full 期间 DW 新写入的数据。
      let cursorAfterJson: string | null = null;
      if (incrementalConfig) {
        const runStartIso = runStart.toISOString();
        const oldCursors: Record<string, string> = JSON.parse(cursorBeforeJson ?? '{}');
        const merged = { ...oldCursors };
        for (const tbl of Object.keys(incrementalConfig.perQuery)) {
          // 防御:cursor 不倒退(老 cursor > runStart 时保留老 — 罕见,跨时钟漂移)
          if (!merged[tbl] || runStartIso > merged[tbl]) {
            merged[tbl] = runStartIso;
          }
        }
        cursorAfterJson = JSON.stringify(merged);
        this.logger.log(`Cursor 写(run_start baseline):${cursorAfterJson}`);
      }
      if (syncLog) {
        const fetched =
          totals.patientsUpserted +
          totals.transactionsWritten +
          totals.duplicates +
          totals.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(),
            cursorAfter: cursorAfterJson,
          },
        });
      }

      // ── 增量空转探针(2026-06-10 游标格式 bug 空转三天才被发现的教训)──
      // fetched=0 且确实带游标跑的增量 → 反查 DW「比游标新的行数」:
      //   DW 也是 0 → 真没新数据,静默;DW > 0 → PAC 拉不到但 DW 有 → 当天告警(企微)。
      const usedCursor =
        incrementalConfig &&
        Object.values(incrementalConfig.perQuery).some((c) => c.cursorValue !== null);
      if (!options.dryRun && usedCursor && totals.transactionsWritten + totals.duplicates === 0 && manifest.sql_source) {
        const counts = await this.chSource.probeNewRowCounts(
          manifest.sql_source,
          incrementalConfig!.perQuery,
        );
        const missed = Object.entries(counts).filter(([, n]) => n > 0);
        if (missed.length > 0) {
          await this.alerter.send({
            level: 'critical',
            title: '增量空转:PAC 拉到 0 行,但 DW 有新数据',
            body:
              `本次增量 fetched=0,但按当前游标反查 DW 仍有未拉取的行 — ` +
              `大概率是游标条件/格式失效(参照 2026-06-10 ISO 游标字典序 bug)。\n` +
              missed.map(([t, n]) => `${t}: ${n} 行`).join('\n'),
            context: { host: host.name, syncLogId: syncLog?.id ?? '' },
          });
        }
      }
    }

    const result: ImportRunResult = {
      runId,
      hostId: host.id,
      hostName: host.name,
      tenantIds: [...seenTenants].sort(),
      dryRun: !!options.dryRun,
      syncLogId: syncLog?.id,
      status,
      totals,
      perResource,
    };

    this.logger.log(
      `Sync 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 patientRelationCfg = assemblerConfigs.find((c) => c.canonical === 'patient_relation');
    const patientReturnVisitCfg = assemblerConfigs.find((c) => c.canonical === 'patient_return_visit');
    const subjectCfgs = assemblerConfigs.filter(
      (c) =>
        c.canonical !== 'patient' &&
        c.canonical !== 'patient_relation' &&
        c.canonical !== 'patient_return_visit',
    );

    // 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`;
    }

    if (patientRelationCfg) {
      const stats = await this.processPatientRelations(
        tables, patientRelationCfg, host.id, tenantResolver, normalize, false,
      );
      perResource.push(stats);
      totals.failed += stats.failed;
    }

    if (patientReturnVisitCfg) {
      const stats = await this.processPatientReturnVisits(
        tables, patientReturnVisitCfg, host.id, tenantResolver, normalize, false,
      );
      perResource.push(stats);
      totals.failed += stats.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,
    };
  }

  // ─────────────────────────────────────────────────────────
  // processCohort —— 单次 load → transform → assemble → write 循环
  //   单 shot 模式:cohort=null,load 全表
  //   cohort 模式:  cohort=patient pairs,load 仅这些 patient 的数据
  //
  // 每次调用结束,tables / transform 中间产物全部出作用域 → V8 GC 释放(降内存峰值)
  // ─────────────────────────────────────────────────────────
  private async processCohort(args: {
    absDir: string;
    manifest: ColdImportManifest;
    host: { id: string; name: string };
    tenantResolver: TenantResolver;
    seenTenants: Set<string>;
    normalize: { amountUnit: 'fen' | 'yuan'; timezone: string };
    patientCfg: AssemblerConfig | undefined;
    patientRelationCfg: AssemblerConfig | undefined;
    patientReturnVisitCfg: AssemblerConfig | undefined;
    subjectCfgs: AssemblerConfig[];
    syncLogId: string | undefined;
    dryRun: boolean;
    cohort: ReadonlyArray<CohortKey> | null;
    incrementalConfig: IncrementalConfig | undefined;
    /// 累加进调用方;由调用方维护(跨 cohort 全 run 汇总)
    perResource: PerResourceStats[];
    totals: TotalsBlock;
    onFirstError: (msg: string) => void;
  }): Promise<void> {
    // 1. Load raw tables(cohort 过滤 / 全表)
    let tables: Record<string, unknown[]>;
    if (args.cohort && args.manifest.sql_source) {
      tables = await this.chSource.loadTablesForCohort(
        args.manifest.sql_source,
        args.cohort,
        args.incrementalConfig,
      );
    } else {
      tables = await this.loadAllTables(args.absDir, args.manifest, args.incrementalConfig);
    }

    // 2. Layer A.5 transforms
    if (args.manifest.transforms && args.manifest.transforms.length > 0) {
      this.logger.log(
        `Layer A.5 transforms:${args.manifest.transforms.length} 步,sources=${Object.keys(tables).length}`,
      );
      const transformInputs = tables as Record<string, Record<string, unknown>[]>;
      tables = this.transformEngine.run({
        tables: transformInputs,
        transforms: args.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(', ')}`,
      );
    }

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

    // 3.5 patient_relation(联系人/亲属边)— 须在 patients 之后(两端解析靠 patient 索引)
    if (args.patientRelationCfg) {
      const stats = await this.processPatientRelations(
        tables,
        args.patientRelationCfg,
        args.host.id,
        args.tenantResolver,
        args.normalize,
        args.dryRun,
      );
      args.perResource.push(stats);
      args.totals.failed += stats.failed;
      this.logger.log(
        `patient_relation: upserted=${stats.patientsUpserted} failed=${stats.failed}`,
      );
    }

    // 3.6 patient_return_visit(诊所回访,展示用)— 同样在 patients 之后(靠 patient 索引解析)
    if (args.patientReturnVisitCfg) {
      const stats = await this.processPatientReturnVisits(
        tables,
        args.patientReturnVisitCfg,
        args.host.id,
        args.tenantResolver,
        args.normalize,
        args.dryRun,
      );
      args.perResource.push(stats);
      args.totals.failed += stats.failed;
      this.logger.log(
        `patient_return_visit: upserted=${stats.patientsUpserted} failed=${stats.failed}`,
      );
    }

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

  /// 读已存在 sync_log 的 metadata.cohortDone(用于 --resume 续跑)
  /// 当前实现:syncLog 是本次新建的 running 行,metadata 为空 → 返回 0
  /// 真正 resume 需 PR3 加 --resume CLI 参数 + 复用已有 running sync_log id(stale 锁释放后)
  private async readCheckpointOffset(syncLogId: string | undefined): Promise<number> {
    if (!syncLogId) return 0;
    const log = await this.prisma.syncLog.findUnique({
      where: { id: syncLogId },
      select: { metadata: true },
    });
    const meta = log?.metadata as { cohortDone?: number } | null;
    return meta?.cohortDone ?? 0;
  }

  // ─────────────────────────────────────────────────────────
  // 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,
        // 专属客服(current_task_director)并入 preferences.dedicatedCs(零迁移,非画像特征)
        preferences: mergePatientPreferences(c),
        // 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 需要从 canonical 拿;host 通过 host_tags / host_notes → assembler.yaml 映射,此处兜底
        // (primaryContactType 已废弃 → 联系人改走 PatientRelation 边表,processPatientRelations)
        tags: Array.isArray(c.tags) ? (c.tags as string[]) : [],
        notes: (c.notes as string | undefined) ?? null,
        // 获客渠道(A.2.1):一级 PAC 标准枚举(assembler enum_mapping 已归一)/ 二级 host 原值
        acquisitionChannel: (c.acquisitionChannel as string | undefined) ?? null,
        acquisitionSub: (c.acquisitionSub as string | undefined) ?? null,
        // 转介绍(B.1.2):recommend_num / recommend_amount(元→分)
        referralCount: c.referralCount != null ? Math.trunc(Number(c.referralCount)) : null,
        referralAmountCents:
          c.referralAmount != null ? Math.round(Number(c.referralAmount) * 100) : null,
      };

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

    return stats;
  }

  // ─────────────────────────────────────────────────────────
  // patient_relation 资源 → upsert 患者-患者关系边(联系人/亲属)
  //   canonical: { patientExternalId, relatedExternalId, relationship }
  //   本人(patientExternalId)必须能解析到 PAC patient(否则无处挂靠 → skip);
  //   关系人(relatedExternalId)解析到 PAC patient 时填 relatedPatientId(给姓名/电话),否则 null。
  // ─────────────────────────────────────────────────────────
  private async processPatientRelations(
    tables: Record<string, unknown[]>,
    config: AssemblerConfig,
    hostId: string,
    tenantResolver: TenantResolver,
    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);
    if (dryRun) return stats;

    // 本租户 externalId → PAC patient id 索引(本人 + 关系人解析共用)
    const indexByTenant = new Map<string, Map<string, string>>();
    const getIndex = async (tenantId: string) => {
      let idx = indexByTenant.get(tenantId);
      if (!idx) {
        idx = await this.buildPatientIndex(hostId, tenantId);
        indexByTenant.set(tenantId, idx);
      }
      return idx;
    };

    for (const { canonical, rawSource } of assembled.rows) {
      const c = canonical as Record<string, unknown>;
      const patientExternalId = String(c.patientExternalId ?? '').trim();
      const relatedExternalId = String(c.relatedExternalId ?? '').trim();
      const relationship = String(c.relationship ?? '').trim();
      if (!patientExternalId || !relatedExternalId || !relationship) {
        stats.failed++;
        continue;
      }
      const tenantId = tenantResolver.resolve(rawSource);
      if (!tenantId) {
        stats.failed++;
        continue;
      }
      const idx = await getIndex(tenantId);
      const patientId = idx.get(patientExternalId);
      if (!patientId) continue; // 本人不在 PAC(非 active client)→ 无处挂靠,跳过
      const relatedPatientId = idx.get(relatedExternalId) ?? null;

      try {
        await this.withDbRetry(
          () =>
            this.prisma.patientRelation.upsert({
              where: {
                patientId_relatedExternalId_relationship: {
                  patientId,
                  relatedExternalId,
                  relationship,
                },
              },
              create: {
                hostId,
                tenantId,
                patientId,
                relatedExternalId,
                relatedPatientId,
                relationship,
              },
              update: { relatedPatientId }, // 关系人后来才入库 → 回填 link
            }),
          'patientRelation upsert',
        );
        stats.patientsUpserted++; // 复用计数字段(resource 标签区分);不并入 totals.patients
      } catch (err) {
        stats.failed++;
        this.logger.error(
          `patientRelation upsert failed ${patientExternalId}→${relatedExternalId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }
    return stats;
  }

  // ─────────────────────────────────────────────────────────
  // patient_return_visit 资源 → upsert 诊所回访记录(展示用,5 试点)
  //   本人(patientExternalId=customer_id)须解析到 PAC patient(否则无处挂靠 → skip)。
  // ─────────────────────────────────────────────────────────
  private async processPatientReturnVisits(
    tables: Record<string, unknown[]>,
    config: AssemblerConfig,
    hostId: string,
    tenantResolver: TenantResolver,
    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);
    if (dryRun) return stats;

    const indexByTenant = new Map<string, Map<string, string>>();
    const getIndex = async (tenantId: string) => {
      let idx = indexByTenant.get(tenantId);
      if (!idx) {
        idx = await this.buildPatientIndex(hostId, tenantId);
        indexByTenant.set(tenantId, idx);
      }
      return idx;
    };

    for (const { canonical, rawSource } of assembled.rows) {
      const c = canonical as Record<string, unknown>;
      const externalId = String(c.externalId ?? '').trim();
      const patientExternalId = String(c.patientExternalId ?? '').trim();
      if (!externalId || !patientExternalId) {
        stats.failed++;
        continue;
      }
      const tenantId = tenantResolver.resolve(rawSource);
      if (!tenantId) {
        stats.failed++;
        continue;
      }
      const idx = await getIndex(tenantId);
      const patientId = idx.get(patientExternalId);
      if (!patientId) continue; // 本人不在 PAC → skip

      const data = {
        clinicId: (c.clinicId as string | undefined) ?? null,
        taskDate: c.taskDate ? new Date(c.taskDate as string) : null,
        type: (c.type as string | undefined) ?? null,
        status: (c.status as string | undefined) ?? null,
        taskStatus: (c.taskStatus as string | undefined) ?? null,
        // 大类·子项 合并值(transforms H);两列皆空时合并出来只剩 " · " 分隔符 → 归一为 null(回访常无治疗项)
        treatmentItems: normalizeMergedItems(c.treatmentItems as string | undefined),
        followContent: (c.followContent as string | undefined) ?? null,
        result: (c.result as string | undefined) ?? null,
      };
      try {
        await this.withDbRetry(
          () =>
            this.prisma.patientReturnVisit.upsert({
              where: {
                hostId_tenantId_externalId: { hostId, tenantId, externalId },
              },
              create: { hostId, tenantId, externalId, patientId, ...data },
              update: { patientId, ...data },
            }),
          'patientReturnVisit upsert',
        );
        stats.patientsUpserted++;
      } catch (err) {
        stats.failed++;
        this.logger.error(
          `patientReturnVisit upsert failed ${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 自己的失败计在内
    stats.mappingMisses = assembled.stats.mappingMisses; // 映射覆盖漏(落 _default 的原值)
    stats.suspectFields = assembled.stats.suspectFields; // 字段/列漂移(整列缺席)
    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`;

    // PR4 ⭐ 批写 buffer — 默认 1000 行一批,createMany + bulkWrite 替代 per-row 写
    //   env PAC_WRITE_BATCH_SIZE 兜底;0 = 退回 per-row(回滚开关)
    const writeBatchSize = resolveWriteBatchSize();

    // 同 batch 内不同 tenant 要分别 createMany(prisma createMany 不能跨 tenant)
    // 简化:按 tenant 分桶 buffer
    const buffersByTenant = new Map<
      string,
      Array<{
        tx: Prisma.PatientTransactionUncheckedCreateInput;
        canonicalRow: Record<string, unknown>;
      }>
    >();

    const flushTenant = async (tenantId: string): Promise<void> => {
      const buf = buffersByTenant.get(tenantId);
      if (!buf || buf.length === 0) return;
      await this.flushBatchedWrite({
        tenantId,
        buffer: buf,
        stats,
        config,
      });
      buf.length = 0;
    };

    try {
      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);
        }
        // stub auto-create
        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;

        // 入 buffer
        let buf = buffersByTenant.get(tenantId);
        if (!buf) {
          buf = [];
          buffersByTenant.set(tenantId, buf);
        }
        buf.push({ tx: txn, canonicalRow });

        // batch 满 → flush 该 tenant
        if (writeBatchSize > 0 && buf.length >= writeBatchSize) {
          await flushTenant(tenantId);
        }
      }

      // 尾巴 — flush 所有未满 buffer
      for (const tenantId of buffersByTenant.keys()) {
        await flushTenant(tenantId);
      }
    } catch (err) {
      this.logger.error(
        `processSubject(${config.canonical}) 批写主循环异常: ${
          err instanceof Error ? err.message : String(err)
        }`,
      );
      throw err;
    }

    return stats;
  }

  /**
   * 带重试的 Prisma 操作 — Postgres/Prisma 引擎偶发瞬时错误:
   *   "Response from the Engine was empty"(引擎 socket 抖动)/ 连接池超时 / ECONNRESET / server closed。
   * 一次全量(260 批 × 多次 SQL)里一次抖动就会整轮 abort(processSubject 主循环 re-throw),
   * 故对热路径 DB 读/写加指数退避重试(默认 3 次:0.5s / 1.5s / 4.5s)。
   * 只重试已知瞬时类错误;唯一约束(P2002)等确定性错误不匹配 → 立即抛(交 caller 处理)。
   */
  private async withDbRetry<T>(
    op: () => Promise<T>,
    label: string,
    maxAttempts = 3,
  ): Promise<T> {
    let lastErr: unknown;
    for (let attempt = 1; attempt <= maxAttempts; attempt++) {
      try {
        return await op();
      } catch (err) {
        lastErr = err;
        const msg = err instanceof Error ? err.message : String(err);
        const transient =
          /Response from the Engine was empty|Can't reach database|connection pool|connection closed|timed out|timeout|ECONNRESET|socket hang up|Connection terminated|server closed the connection|reset by peer|too many connections/i.test(
            msg,
          );
        if (!transient || attempt === maxAttempts) throw err;
        const backoffMs = 500 * Math.pow(3, attempt - 1); // 500 / 1500 / 4500
        this.logger.warn(
          `[db] "${label}" 第 ${attempt}/${maxAttempts} 次失败(瞬时:${msg.slice(0, 80)}),${backoffMs}ms 后重试`,
        );
        await new Promise((r) => setTimeout(r, backoffMs));
      }
    }
    throw lastErr;
  }

  /**
   * PR4 批写一批 transactions + 衍生 facts。
   *
   * 流程:
   *   1. prisma.patientTransaction.createMany({ data, skipDuplicates: true })
   *      → 1 次 SQL 写 N 行,撞 source_event_id 唯一约束自动 skip
   *   2. SELECT ... WHERE sourceEventId IN (...) 取回所有 tx 行(含 id)— 1 次 SQL
   *      新增 + 已存在的都会回来,parser 全部跑(idempotent via fact hash)
   *   3. parserPipeline.runForBatch(items) → 内部 bulkWrite,1 次 SELECT 现有 fact + 批 INSERT/UPDATE
   *
   * 相比 per-row 老路径:N 行从 ~3N 次 SQL 降到 ~5 次 SQL → 10-20x 速度
   */
  private async flushBatchedWrite(args: {
    tenantId: string;
    buffer: Array<{
      tx: Prisma.PatientTransactionUncheckedCreateInput;
      canonicalRow: Record<string, unknown>;
    }>;
    stats: PerResourceStats;
    config: AssemblerConfig;
  }): Promise<void> {
    const { tenantId, buffer, stats, config } = args;
    if (buffer.length === 0) return;
    const hostId = buffer[0]!.tx.hostId as string;

    // 1. createMany tx(skipDuplicates 跳重)
    const data = buffer.map((b) => b.tx);
    let createdCount = 0;
    try {
      const r = await this.withDbRetry(
        () =>
          this.prisma.patientTransaction.createMany({
            data,
            skipDuplicates: true,
          }),
        'createMany tx',
      );
      createdCount = r.count;
    } catch (err) {
      // createMany 整批失败(罕见 — schema 校验等)→ 降级 per-row
      this.logger.warn(
        `createMany tx 批失败,降级 per-row: ${err instanceof Error ? err.message : String(err)}`,
      );
      await this.fallbackPerRowWrite(buffer, stats, config);
      return;
    }
    stats.transactionsWritten += createdCount;
    stats.duplicates += buffer.length - createdCount;

    // 2. 取回所有 tx 行(含 id),供 parser 用
    const eventIds = buffer
      .map((b) => b.tx.sourceEventId as string | undefined)
      .filter((id): id is string => !!id);
    if (eventIds.length === 0) {
      return; // 全 buffer 都没 source_event_id?理论不该发生
    }
    const txRows = await this.withDbRetry(
      () =>
        this.prisma.patientTransaction.findMany({
          where: {
            hostId,
            tenantId,
            sourceEventId: { in: eventIds },
          },
          select: {
            id: true,
            hostId: true,
            tenantId: true,
            patientId: true,
            action: true,
            subjectType: true,
            subjectId: true,
            occurredAt: true,
            clinicId: true,
            sourceEventId: true,
          },
        }),
      'findMany tx by sourceEventId',
    );
    const txBySourceEventId = new Map(txRows.map((t) => [t.sourceEventId!, t]));

    // 3. 构建 batch items,跑 parserPipeline.runForBatch
    const items: BatchItem[] = [];
    for (const b of buffer) {
      const sid = b.tx.sourceEventId as string | undefined;
      if (!sid) continue;
      const tx = txBySourceEventId.get(sid);
      if (!tx) {
        // 罕见 — createMany 写入但 SELECT 没拿到(并发?最大可见性问题?)
        this.logger.warn(`flushBatch: tx ${sid} 写入后未能查回(跳过 parser)`);
        continue;
      }
      items.push({
        transaction: {
          id: tx.id,
          hostId: tx.hostId,
          tenantId: tx.tenantId,
          patientId: tx.patientId,
          action: tx.action as Action,
          subjectType: tx.subjectType,
          subjectId: tx.subjectId,
          occurredAt: tx.occurredAt,
          clinicId: tx.clinicId,
        },
        canonicalRow: b.canonicalRow,
      });
    }
    if (items.length === 0) return;

    const metrics = await this.parserPipeline.runForBatch(items);
    stats.factsCreated += metrics.factsCreated;
    stats.factsSuperseded += metrics.factsSuperseded;
    stats.factsUnchanged += metrics.factsUnchanged;
    stats.factsEvidenceAppended += metrics.factsEvidenceAppended;
    stats.factsFailed += metrics.factsFailed;
  }

  /// 兜底:createMany 整批失败时降级 per-row(慢但稳)
  private async fallbackPerRowWrite(
    buffer: Array<{
      tx: Prisma.PatientTransactionUncheckedCreateInput;
      canonicalRow: Record<string, unknown>;
    }>,
    stats: PerResourceStats,
    config: AssemblerConfig,
  ): Promise<void> {
    for (const b of buffer) {
      const txn = b.tx;
      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: b.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++;
          const existing = await this.prisma.patientTransaction.findFirst({
            where: {
              hostId: txn.hostId as string,
              tenantId: txn.tenantId as string,
              sourceEventId: (txn.sourceEventId as string | undefined) ?? 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: b.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(
            `fallback per-row tx failed (resource=${config.canonical}, subject=${txn.subjectId}): ${
              err instanceof Error ? err.message : String(err)
            }`,
          );
        }
      }
    }
  }

  private async buildPatientIndex(
    hostId: string,
    tenantId: string,
  ): Promise<Map<string, string>> {
    const rows = await this.withDbRetry(
      () =>
        this.prisma.patient.findMany({
          where: { hostId, tenantId },
          select: { id: true, externalId: true },
        }),
      'buildPatientIndex',
    );
    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.withDbRetry(
      () =>
        this.prisma.patient.upsert({
          where: { hostId_tenantId_externalId: { hostId, tenantId, externalId } },
          create: { hostId, tenantId, externalId, active: true },
          update: {}, // 有则 noop(姓名等真实主档 upsert 走 processPatients 路径)
        }),
      'ensurePatientStub patient',
    );
    await this.withDbRetry(
      () =>
        this.prisma.patientProfile.upsert({
          where: { patientId: patient.id },
          create: { patientId: patient.id, doNotContact: false, deceased: false, tags: [] },
          update: {},
        }),
      'ensurePatientStub profile',
    );
    return patient.id;
  }

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

  /// 扫 dataDir 下各 host 子目录的 manifest,返回 `auto_sync: true` 的【目录名】列表。
  /// 给 SyncIncrementalSchedulerService 自动发现"该每晚自动拉的 host" —— 取代 env 清单,
  /// 让"是否自动同步"跟 host 配置同处声明,接入新 host 不必再改 env(防 drift)。
  /// 宽松:无 manifest / 解析失败 / auto_sync ≠ true 的目录跳过(完整校验在 importDirectory 时做)。
  discoverAutoSyncHostDirs(dataDir: string): string[] {
    if (!fs.existsSync(dataDir)) return [];
    const out: string[] = [];
    for (const entry of fs.readdirSync(dataDir, { withFileTypes: true })) {
      if (!entry.isDirectory()) continue;
      const mPath = path.join(dataDir, entry.name, 'manifest.yaml');
      if (!fs.existsSync(mPath)) continue;
      try {
        const parsed = yaml.load(fs.readFileSync(mPath, 'utf-8')) as
          | { auto_sync?: unknown }
          | undefined;
        if (parsed?.auto_sync === true) out.push(entry.name);
      } catch {
        // 解析失败跳过 — 不阻塞其他 host 的发现
      }
    }
    return out.sort();
  }

  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;
  }

  /// 读最近一次成功的 sync_log 的 cursor_after JSON。
  /// 返回 { tableName: cursorValue } map(无记录返回空对象 → 首跑等价全量)。
  /// PR3 统一:resource = 'incremental_bundle';历史 'cold_import_bundle' 行无 cursor_after,
  /// 自然被 cursorAfter: { not: null } 过滤掉(不影响)。
  private async readLastIncrementalCursor(hostId: string): Promise<Record<string, string>> {
    const last = await this.prisma.syncLog.findFirst({
      where: {
        hostId,
        resource: 'incremental_bundle',
        // SUCCESS or PARTIAL 都接受(partial 表示有进展,cursor 已推进到 run_start)
        status: { in: [SyncStatus.SUCCESS, SyncStatus.PARTIAL] },
        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: [],
      mappingMisses: [],
      suspectFields: [],
    };
  }
}

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

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

/// 沿 transforms 的 output→input 链回溯,找到非任何 transform 产出的"原始源表"名。
/// 例:diagnosis_rows ← _diagnosis_norm ← _diagnosis_raw ← fact_emr_treatment_out(终点=源表)。
/// 用于 reparse:transaction.rawPayload 存的就是这张源表的行。
function traceRawSourceTable(primaryTable: string, transforms: ReadonlyArray<unknown>): string {
  const byOutput = new Map<string, string>(); // output 表名 → input 表名
  for (const tf of transforms) {
    const t = tf as { input?: string; output?: string; outputs?: Array<{ output?: string }> };
    if (t.output && t.input) byOutput.set(t.output, t.input);
    // route_by_pattern 多 output:每个 output 都回到同一 input
    if (Array.isArray(t.outputs) && t.input) {
      for (const o of t.outputs) if (o?.output) byOutput.set(o.output, t.input);
    }
  }
  let cur = primaryTable;
  const seen = new Set<string>();
  while (byOutput.has(cur) && !seen.has(cur)) {
    seen.add(cur);
    cur = byOutput.get(cur)!;
  }
  return cur;
}

export interface PerResourceStats extends TotalsBlock {
  resource: string;
  fetched: number;
  sampleCanonical: unknown[];
  /// 映射覆盖漏(assembler 透传):该资源装配时落 _default 的原值(field+rawValue 聚合)
  mappingMisses: MappingMiss[];
  /// 字段/列漂移(assembler 透传):整列缺席的映射字段(疑似宿主改名/删列)
  suspectFields: SuspectField[];
}

/**
 * cohort batch size 解析:
 *   option 显式 > 0 → 用 option
 *   option 显式 0/null → 强制 single-shot(向后兼容,文件源也走此路径)
 *   option 未指定 → fallback env PAC_COHORT_BATCH_SIZE(默认 5000)
 *
 * 返回 0 = 不分批(single-shot);>0 = cohort 大小
 */
export function resolveCohortBatchSize(opt: number | null | undefined): number {
  if (opt !== undefined && opt !== null) {
    return opt > 0 ? Math.floor(opt) : 0;
  }
  const env = process.env.PAC_COHORT_BATCH_SIZE;
  if (env !== undefined) {
    const n = parseInt(env, 10);
    return Number.isFinite(n) && n > 0 ? n : 0;
  }
  return 5000;
}

/**
 * PR4 ⭐ batch size for createMany tx + bulk fact write。
 *   env PAC_WRITE_BATCH_SIZE 优先(0 = 退回 per-row,回滚开关)
 *   默认 1000(SQL prepared-statement 上限 + V8 array 友好,实测平衡)
 */
export function resolveWriteBatchSize(): number {
  const env = process.env.PAC_WRITE_BATCH_SIZE;
  if (env !== undefined) {
    const n = parseInt(env, 10);
    return Number.isFinite(n) && n >= 0 ? n : 0;
  }
  return 1000;
}

/**
 * PR5b ⭐ cohort 并发度:
 *   env PAC_COHORT_CONCURRENCY > 1 → 并行处理 N 个 cohort(worker pool)
 *   默认 1 = 完全串行(行为不变,最稳)
 *
 * ⚠️ 并行时注意:
 *   - 数据完整性不受影响(cohort = disjoint 患者;source_event_id + fact version partial UNIQUE 双保险)
 *   - checkpoint resume 禁用(完成乱序,靠幂等从头重跑)
 *   - 需 bump Prisma 连接池:DATABASE_URL?connection_limit=N×(write_batch 并发数+余量),
 *     否则并发 createMany/bulkWrite 排队等连接(不损坏数据,只是变慢)
 *   - 内存 N× 单 cohort(单 cohort ~330MB;并发 4 ≈ 1.3GB,14GB 机器有余量)
 */
export function resolveCohortConcurrency(): number {
  const env = process.env.PAC_COHORT_CONCURRENCY;
  if (env !== undefined) {
    const n = parseInt(env, 10);
    return Number.isFinite(n) && n >= 1 ? n : 1;
  }
  return 1;
}

/** 把数组切成 size 大小的块(尾巴块可能 < size) */
export function chunk<T>(arr: ReadonlyArray<T>, size: number): T[][] {
  if (size <= 0) return [arr.slice() as T[]];
  const out: T[][] = [];
  for (let i = 0; i < arr.length; i += size) {
    out.push(arr.slice(i, i + size) as T[]);
  }
  return out;
}

/**
 * 给 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}`;
}

/// 增量游标 → DW 字符串列可比的字面量。
/// DW 的 cursor 列是 String('YYYY-MM-DD HH:mm:ss[.SSS]',宿主本地时区)→ 比较走字典序;
/// 游标存的是 run_start ISO(带 'T'/'Z')→ 'T' 比 ' ' 字典序大,所有真实行都排在游标前
/// → fetched 永远 0。这里把 ISO 转成 DW 时区的 'YYYY-MM-DD HH:mm:ss'(秒精度,不带毫秒:
/// 同秒带 .SSS 的行值字典序更大 → 仍会被 > 选中,轻微重叠由 source_event_id 幂等吃掉)。
/// 已是普通格式的值(历史行值推进的游标)原样返回。
function toDwCursorLiteral(value: string, timeZone: string): string {
  if (!/\d{4}-\d{2}-\d{2}T/.test(value)) return value;
  const d = new Date(value);
  if (Number.isNaN(d.getTime())) return value;
  // sv-SE locale 恰好输出 'YYYY-MM-DD HH:mm:ss'
  return new Intl.DateTimeFormat('sv-SE', {
    timeZone,
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
    hour: '2-digit',
    minute: '2-digit',
    second: '2-digit',
    hour12: false,
  }).format(d);
}

/// 回访治疗项「大类 · 子项」合并值归一(transforms H 的 concat)。
/// 两列皆空 → 合并出来只剩 " · "(纯分隔符/空白)→ 视为无治疗项,归 null;有真实内容则 trim 返回。
function normalizeMergedItems(v: string | undefined): string | null {
  if (!v) return null;
  const trimmed = v.trim();
  // 去掉所有分隔符/空白后仍有内容才保留(避免 "·" / " · " / "" 这类空壳入库)
  return trimmed.replace(/[·\s/]+/g, '') ? trimmed : null;
}

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[];
}

/**
 * 并发拦截 — 同 host 已有 status='running' 的 sync_log 时抛出。
 * 由 sync_logs 的 partial UNIQUE (host_id) WHERE status='running' 约束触发(P2002)。
 *
 * 调用方处理建议:
 *   - cron(SyncIncrementalScheduler):捕获本异常 → warn log + skip + 等下次 cron
 *   - 手动 CLI:打印 error + 退出 code 2(让运维知道有其他 sync 在跑)
 *   - 单患者 refresh:本异常不会抛(走 resource='patient_refresh' 不参与该锁)
 */
export class SyncAlreadyRunningError extends Error {
  constructor(
    public readonly hostName: string,
    public readonly existing: { id: string; startedAt: Date; triggeredBy: string | null } | null,
  ) {
    super(
      `SYNC_ALREADY_RUNNING:host=${hostName}` +
        (existing
          ? ` existing=${existing.id} started_at=${existing.startedAt.toISOString()} triggered_by=${existing.triggeredBy}`
          : ''),
    );
    this.name = 'SyncAlreadyRunningError';
  }
}
