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 } 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 { 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;
      /// 按 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, cursorValue: lastCursor[table] ?? 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 subjectCfgs = assemblerConfigs.filter((c) => c.canonical !== 'patient');
      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,
              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,
          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,
          },
        });
      }
    }

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

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

    // 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,
        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.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;
  }

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

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

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

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

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