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

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

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

  async importDirectory(
    dir: string,
    options: { dryRun?: boolean; incremental?: boolean } = {},
  ): Promise<ImportRunResult> {
    const absDir = path.resolve(dir);
    const runId = randomUUID();
    // ⭐ 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 准备(必须在 SyncLog 创建之前算,因为 cursorBefore 要落到 sync_log 行)
    let incrementalConfig: IncrementalConfig | undefined;
    let cursorBeforeJson: string | null = null;
    if (options.incremental) {
      const perQueryCfg = manifest.sql_source?.incremental?.per_query;
      if (!perQueryCfg) {
        throw new Error(
          `--incremental 模式但 manifest.sql_source.incremental.per_query 未配置;请在 manifest.yaml 补 cursor 配置`,
        );
      }
      const lastCursor = await this.readLastIncrementalCursor(host.id);
      this.logger.log(
        `Incremental cursor 读:${Object.keys(lastCursor).length === 0 ? '(首跑,全量)' : JSON.stringify(lastCursor)}`,
      );
      incrementalConfig = {
        perQuery: Object.fromEntries(
          Object.entries(perQueryCfg).map(([table, cfg]) => [
            table,
            { cursorColumn: cfg.cursor_column, cursorValue: lastCursor[table] ?? null },
          ]),
        ),
      };
      cursorBeforeJson = JSON.stringify(
        Object.fromEntries(
          Object.entries(incrementalConfig.perQuery).map(([k, v]) => [k, v.cursorValue ?? '']),
        ),
      );
    }

    // 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';
    const syncResource = options.incremental ? 'incremental_bundle' : 'cold_import_bundle';
    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: `${options.incremental ? 'incremental' : 'cold_import'}:${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. 一次性加载所有 raw tables(文件 / ClickHouse 二选一)
      let tables = await this.loadAllTables(absDir, manifest, incrementalConfig);

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

      // 6. 一次性读所有 assembler configs
      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 };

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

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

      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,无害)
      let cursorAfterJson: string | null = null;
      if (options.incremental && 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(`Incremental 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,
    };
  }

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

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

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

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

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

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

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

    return stats;
  }

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

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

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

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

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

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

    return stats;
  }

  private async buildPatientIndex(
    hostId: string,
    tenantId: string,
  ): Promise<Map<string, string>> {
    const rows = await this.prisma.patient.findMany({
      where: { hostId, tenantId },
      select: { id: true, externalId: true },
    });
    return new Map(rows.map((p) => [p.externalId, p.id]));
  }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

/**
 * 并发拦截 — 同 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';
  }
}
