import { Injectable, Logger } from '@nestjs/common';
import { createClient, type ClickHouseClient } from '@clickhouse/client';
import type { ClickHouseSource } from './manifest.schema';

/**
 * ClickHouseSourceService — 把 ClickHouse SQL 查询结果作为 cold-import 数据源。
 *
 * 形态:执行 manifest.sql_source.queries 里每个 SQL,把结果以
 *      `Record<tableName, row[]>` 形式返回,跟文件源完全等价
 *      (下游 AssemblerEngine 无感知差别)。
 *
 * 短期 cold-import 直连模式;W4+ 升级为带 cursor 的真增量 pull(ClickHouseSqlPullStrategy)。
 *
 * 安全:密码从环境变量读取(password_env),plaintext 永不进 yaml。
 */
/// W4 末:DW 增量配置(per table cursor)
export interface IncrementalConfig {
  /// per query 配置:cursor 列名(host 表那边的字段)+ 上次 cursor 值(从 sync_logs 读)
  perQuery: Record<string, { cursorColumn: string; cursorValue: string | null }>;
  /// 跑完后返回新的 cursor 值(每张表的 max)
  cursorAdvances?: Record<string, string>;
}

/// Cohort batch 的患者 key — key=患者主键值,tenant=租户区分值(可选,仅配了 tenant_key_column)
/// 列名由 manifest.sql_source.cohort 声明,本接口只承载值(宿主无关)
export interface CohortKey {
  key: string;
  tenant?: string;
}

/// W4 末:单患者刷新 scope(详情页"刷新"按钮)
/// 把 manifest queries 的 cohort 子查询替换为 `WHERE patient_id='X' AND brand='Y'`
/// 不跑 cursor、不写 cursor、不反向拉主档(本身就是单患者全量)
export interface PatientScope {
  /// host 侧 patient_id(字符串,DW patient_id 是 String 类型)
  patientExternalId: string;
  /// host 侧 brand(中文 "瑞尔" / "瑞泰" — DW 真实字段值)
  brand: string;
}

@Injectable()
export class ClickHouseSourceService {
  private readonly logger = new Logger(ClickHouseSourceService.name);

  /**
   * 带重试的 CH 查询 — 远程 DW(阿里云 ADS 公网出口)偶发瞬时错误:
   *   "Response from the Engine was empty" / 连接 reset / 超时。
   * 一次性大全量(260 批 × 6 query = 1560 次)里只要一次抖动就整跑失败,
   * 故对每个 query 加指数退避重试(默认 3 次:0.5s / 1.5s / 4.5s)。
   *
   * 只重试"瞬时"类错误;SQL 语法错等确定性错误不重试(立即抛)。
   */
  private async queryJsonWithRetry(
    client: ClickHouseClient,
    sql: string,
    label: string,
    maxAttempts = 3,
  ): Promise<unknown[]> {
    let lastErr: unknown;
    for (let attempt = 1; attempt <= maxAttempts; attempt++) {
      try {
        const result = await client.query({ query: sql, format: 'JSONEachRow' });
        return (await result.json()) as unknown[];
      } catch (err) {
        lastErr = err;
        const msg = err instanceof Error ? err.message : String(err);
        // 确定性错误(SQL 错 / 权限)不重试 — 重试也没用,快速失败。
        // ⚠️ ETIMEDOUT 不含 "timeout" 子串 —— 必须显式列(高并发打远程 DW 时 socket read 超时,
        //    曾导致整轮 fatal abort 而非重试)。一并补 socket 级网络错误码 + CH client "HTTP request error"。
        const transient =
          /empty|timeout|timed out|ETIMEDOUT|ECONNRESET|ECONNREFUSED|EHOSTUNREACH|ENETUNREACH|EPIPE|socket|network|HTTP request error|fetch failed|EAI_AGAIN|503|502|too many|reset by peer/i.test(
            msg,
          );
        if (!transient || attempt === maxAttempts) {
          throw err;
        }
        const backoffMs = 500 * Math.pow(3, attempt - 1); // 500 / 1500 / 4500
        this.logger.warn(
          `[clickhouse] "${label}" 第 ${attempt}/${maxAttempts} 次失败(瞬时:${msg.slice(0, 80)}),${backoffMs}ms 后重试`,
        );
        await new Promise((r) => setTimeout(r, backoffMs));
      }
    }
    throw lastErr;
  }

  /// W4 末:env 覆盖 manifest connection(便于多环境部署 不必改 yaml)
  /// 优先级:env → manifest → 抛错
  private resolveConnection(source: ClickHouseSource): {
    url: string;
    database: string;
    username: string;
    password: string;
  } {
    const url = process.env.DW_CLICKHOUSE_URL?.trim() || source.connection.url;
    const database = process.env.DW_CLICKHOUSE_DATABASE?.trim() || source.connection.database;
    const username = process.env.DW_CLICKHOUSE_USERNAME?.trim() || source.connection.username;
    const password = process.env[source.connection.password_env];
    if (!password) {
      throw new Error(
        `ClickHouse 密码缺失:环境变量 ${source.connection.password_env} 未设置;` +
          `请检查 .env 或 dotenv 加载`,
      );
    }
    return { url, database, username, password };
  }

  /// 增量空转探针:按(已规范化的)游标去 DW 数"比游标新的行数"。
  /// 用途:增量 fetched=0 时分辨「DW 真没新数据(静默)」vs「查询条件失效在空转(告警)」
  /// —— 2026-06-10 游标 ISO 格式 bug 让增量空转三天才被人工发现,此探针让它当天就暴露。
  /// 探针绝不能影响同步主流程:任何异常吞掉返回 {}。
  async probeNewRowCounts(
    source: ClickHouseSource,
    perQuery: IncrementalConfig['perQuery'],
  ): Promise<Record<string, number>> {
    const out: Record<string, number> = {};
    try {
      const conn = this.resolveConnection(source);
      const client = createClient({
        url: conn.url,
        database: conn.database,
        username: conn.username,
        password: conn.password,
        request_timeout: 30_000,
      });
      try {
        for (const [tableName, cfg] of Object.entries(perQuery)) {
          if (!cfg.cursorValue) continue;
          const sql = source.queries[tableName];
          const m = sql?.match(/FROM\s+([\w.]+)/i);
          if (!m) continue;
          const q = `SELECT count(*) AS c FROM ${m[1]} WHERE ${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`;
          const rs = await client.query({ query: q, format: 'JSONEachRow' });
          const rows = (await rs.json()) as Array<{ c: string | number }>;
          out[tableName] = Number(rows[0]?.c ?? 0);
        }
      } finally {
        await client.close();
      }
    } catch (e) {
      this.logger.warn(`[probe] 空转探针失败(忽略): ${e instanceof Error ? e.message : String(e)}`);
    }
    return out;
  }

  async loadAllTables(
    source: ClickHouseSource,
    incremental?: IncrementalConfig,
  ): Promise<Record<string, unknown[]>> {
    const conn = this.resolveConnection(source);

    const client: ClickHouseClient = createClient({
      url: conn.url,
      database: conn.database,
      username: conn.username,
      password: conn.password,
      // Aliyun ADS 公网偶尔抽风,加大重试
      request_timeout: 60_000,
      compression: { response: true, request: false },
    });

    const tables: Record<string, unknown[]> = {};
    const defaultLimit = source.default_limit ?? 100_000;

    try {
      // ⭐ PR5a:N 个 query 并行(同 loadTablesForCohort)
      if (incremental) incremental.cursorAdvances = incremental.cursorAdvances ?? {};
      await Promise.all(
        Object.entries(source.queries).map(async ([tableName, sql]) => {
          // W4 末:incremental 模式注入 cursor 条件 + cohort LIMIT 移除(增量不需要 dev cohort)
          const incCfg = incremental?.perQuery[tableName];
          const sqlWithIncremental = incCfg
            ? this.injectIncrementalCursor(sql, incCfg.cursorColumn, incCfg.cursorValue)
            : sql;
          const sqlWithLimit = this.applyDefaultLimit(sqlWithIncremental, defaultLimit);
          const started = Date.now();
          const rows = await this.queryJsonWithRetry(client, sqlWithLimit, tableName);
          const elapsed = Date.now() - started;
          tables[tableName] = rows;
          this.logger.log(
            `[clickhouse]   "${tableName}" → ${rows.length} 行,${elapsed} ms`,
          );
          // W4 末:incremental 跑完算新 cursor(= max(cursor_column))
          if (incremental?.cursorAdvances && incCfg && rows.length > 0) {
            const maxVal = this.computeMax(rows as Record<string, unknown>[], incCfg.cursorColumn);
            if (maxVal) {
              incremental.cursorAdvances[tableName] = maxVal;
            }
          }
        }),
      );

      // ⭐ W4 末 反向拉主档(方案 C):增量模式下,fact 表拉来的 patient_id 不一定在 fact_client_out 里
      //   (例:患者 EMR 编辑了 last_visit_time 不变,主档 cursor 拉不到他)
      //   → 收集所有 fact 表中出现的 patient_id,追加一次 fact_client_out WHERE patient_id IN (...)
      //   合并入已有 fact_client_out tables(set 去重)→ 数据完整性 100%
      //   stub 兜底(方案 A,parser 侧)+ 主档反向拉(方案 C,sync 侧)= 完整双保险
      if (incremental && tables['fact_client_out']) {
        const reverseRows = await this.reversePullPatientMaster(client, tables);
        if (reverseRows.length > 0) {
          const before = tables['fact_client_out'].length;
          // 去重 by patient_id+brand:已有 cursor 拉到的不重复 push
          const existing = new Set(
            (tables['fact_client_out'] as Record<string, unknown>[]).map(
              (r) => `${r.patient_id}|${r.brand}`,
            ),
          );
          for (const r of reverseRows) {
            const key = `${(r as Record<string, unknown>).patient_id}|${(r as Record<string, unknown>).brand}`;
            if (!existing.has(key)) {
              (tables['fact_client_out'] as Record<string, unknown>[]).push(r as Record<string, unknown>);
              existing.add(key);
            }
          }
          this.logger.log(
            `[clickhouse]   反向拉主档:+${tables['fact_client_out'].length - before} 行(原 ${before} → 现 ${tables['fact_client_out'].length})`,
          );
        }
      }
    } finally {
      await client.close();
    }

    return tables;
  }

  /// W4 末:从已拉的 fact 表里收集 patient_id 集合,反向拉对应主档
  /// 必要因为 fact_client_out 用 last_visit_time cursor — patient 主档 PII 没变但 fact 变了(EMR 编辑)
  ///   →  fact 表带的 patient_id 不在 cursor 拉的主档里 → stub 派生,但 PII 缺失
  /// 反向拉补齐:确保所有 fact 引用的 patient 都有真实主档(姓名/电话/性别等)
  private async reversePullPatientMaster(
    client: ClickHouseClient,
    tables: Record<string, unknown[]>,
  ): Promise<unknown[]> {
    // 收集所有 fact 表的 patient_id + brand 集合(去重)
    const factTables = ['fact_appointment_out', 'fact_emr_treatment_out', 'fact_settlement_out', 'fact_settlement_mode_out'];
    const pairs = new Set<string>();
    for (const t of factTables) {
      for (const row of (tables[t] ?? []) as Record<string, unknown>[]) {
        const pid = row.patient_id;
        const brand = row.brand;
        if (pid !== null && pid !== undefined && brand) {
          pairs.add(`${pid}|||${brand}`);
        }
      }
    }
    if (pairs.size === 0) return [];
    // 构造 IN ((pid1,brand1),(pid2,brand2),...) tuple list
    // CH 支持 `(patient_id, brand) IN ((1,'瑞尔'),(2,'瑞泰'))` tuple in 语法
    const tuples = [...pairs]
      .map((p) => {
        const [pid, brand] = p.split('|||');
        return `('${(pid ?? '').replace(/'/g, "''")}', '${(brand ?? '').replace(/'/g, "''")}')`;
      })
      .join(',');
    const sql = `SELECT * FROM dw_group.fact_client_out WHERE (patient_id, brand) IN (${tuples})`;
    this.logger.log(`[clickhouse] 反向拉主档 query — ${pairs.size} pairs`);
    const started = Date.now();
    const result = await client.query({ query: sql, format: 'JSONEachRow' });
    const rows = (await result.json()) as unknown[];
    this.logger.log(`[clickhouse]   反向 fact_client_out → ${rows.length} 行,${Date.now() - started} ms`);
    return rows;
  }

  /// W4 末:单患者刷新 — 把 manifest queries 全部改写为 `WHERE patient_id='X' AND brand='Y'` 范围拉
  ///
  /// 跟 loadAllTables 的差别:
  ///   - 不读 / 不写 sync_logs cursor(daily incremental cron 完全不受影响)
  ///   - 不跑反向拉主档(本身已 scope 到该 patient,主档 fact_client_out 同样 WHERE 拉到)
  ///   - 不应用 default_limit(单患者数据量极小,几十~几百行)
  ///
  /// SQL 改写策略(等价 injectIncrementalCursor 套路):
  ///   1. parse SELECT cols + FROM table
  ///   2. extract 业务过滤(去掉 cohort 子查询、cursor、ORDER BY、LIMIT)
  ///   3. rebuild: `SELECT cols FROM table WHERE patient_id='X' AND brand='Y' [AND <business>]`
  ///   4. fact_client_out 没 patient_id?它**自己就是 patient 主档表**,patient_id 字段就是主键 →
  ///      WHERE 同样过滤即可
  async loadTablesForPatient(
    source: ClickHouseSource,
    scope: PatientScope,
  ): Promise<Record<string, unknown[]>> {
    const conn = this.resolveConnection(source);
    const client: ClickHouseClient = createClient({
      url: conn.url,
      database: conn.database,
      username: conn.username,
      password: conn.password,
      request_timeout: 60_000,
      compression: { response: true, request: false },
    });

    const tables: Record<string, unknown[]> = {};
    const pidEsc = scope.patientExternalId.replace(/'/g, "''");
    const brandEsc = scope.brand.replace(/'/g, "''");

    try {
      // 并行查 N 张源表(单患者数据量极小,瓶颈是跨洋 RTT)。
      // 串行 6 表 ≈ 6×RTT;并行 ≈ 1×RTT。对齐 loadAllTables/loadTablesForCohort 的 Promise.all 口径。
      await Promise.all(
        Object.entries(source.queries).map(async ([tableName, sql]) => {
          const patientSql = this.injectPatientFilter(sql, pidEsc, brandEsc);
          const started = Date.now();
          const result = await client.query({ query: patientSql, format: 'JSONEachRow' });
          const rows = (await result.json()) as unknown[];
          tables[tableName] = rows;
          this.logger.log(
            `[clickhouse·patient]   "${tableName}" → ${rows.length} 行,${Date.now() - started} ms`,
          );
        }),
      );
    } finally {
      await client.close();
    }
    return tables;
  }

  /**
   * 列出全部 patient cohort key,作为 cohort batch 的 ORDER 边界。
   *
   * ⭐ 宿主无关:表名 / 主键列 / 租户列 / cursor 列全部从 manifest.sql_source.cohort 读,
   *   代码不硬编码任何 jvs-dw 专属名称。各 host 改 yaml 即可。
   *
   * SQL 形态:
   *   SELECT DISTINCT <patient_key>[, <tenant_key>]
   *   FROM <patient_list_from>
   *   [WHERE <list_cursor_column> > '<cursor>']
   *   ORDER BY <patient_key>   ← 排序保证 chunk 分页稳定
   *
   * 返回 CohortKey[]:{ key, tenant? }(tenant 仅当配了 tenant_key_column)
   */
  async listPatientPairs(
    source: ClickHouseSource,
    incremental?: IncrementalConfig,
    clinics?: string[],
    since?: string,
  ): Promise<CohortKey[]> {
    const cohort = source.cohort;
    if (!cohort) {
      throw new Error(
        'listPatientPairs 需要 manifest.sql_source.cohort 配置(patient_list_from / patient_key_column 等)',
      );
    }
    const { patient_list_from, patient_key_column, tenant_key_column, list_cursor_column } = cohort;

    const conn = this.resolveConnection(source);
    const client = createClient({
      url: conn.url,
      database: conn.database,
      username: conn.username,
      password: conn.password,
      request_timeout: 60_000,
      compression: { response: true, request: false },
    });
    try {
      // 增量 cursor:patient 列表表的 cursor 值(从 perQuery 拿,key = patient_list_from 表名)
      // 表名可能含库前缀(dw_group.fact_client_out),perQuery key 是不带库前缀的(fact_client_out)
      const listTableShort = patient_list_from.includes('.')
        ? patient_list_from.split('.').pop()!
        : patient_list_from;
      const cursorCfg = incremental?.perQuery[listTableShort];
      const cursorCol = list_cursor_column ?? cursorCfg?.cursorColumn;
      // ⭐ 增量列患者 = 「任一源表有 > 各自游标的行」的患者并集(UNION ALL,manifest 驱动)。
      //   只看 patient_list_from.last_visit_time 会漏"没来就诊但数据有变"的患者 ——
      //   典型:客服电话约了复诊(预约 updated_date 变了,人没来)、医生回头补写病历;
      //   且等患者真来就诊时游标已越过那些行 → 永久漏数。
      //   2026-06-11 服务器实测:增量 fetched=0 而 DW 有 24 行预约 + 3 行 EMR 超游标,
      //   逐行核验 27/27 全部属于"无新就诊"患者(哨兵空转告警由此而来)。
      const incBranches: string[] = [];
      if (incremental) {
        for (const [tbl, cfg] of Object.entries(incremental.perQuery)) {
          if (!cfg.cursorColumn || !cfg.cursorValue) continue;
          const m = source.queries[tbl]?.match(/FROM\s+([\w.]+)/i);
          const fqn = m?.[1] ?? tbl;
          const keyCols = tenant_key_column
            ? `${patient_key_column}, ${tenant_key_column}`
            : patient_key_column;
          incBranches.push(
            `SELECT ${keyCols} FROM ${fqn} WHERE ${cfg.cursorColumn} > '${cfg.cursorValue.replace(/'/g, "''")}'`,
          );
        }
      }
      const unionMode = incBranches.length > 0;
      const whereParts: string[] = [];
      if (!unionMode && cursorCol && cursorCfg?.cursorValue) {
        whereParts.push(`${cursorCol} > '${cursorCfg.cursorValue.replace(/'/g, "''")}'`);
      }
      // dev/ops:只摄入指定患者(PAC_COHORT_ONLY_PATIENT=<patient_id> 或逗号列表 id1,id2,...)
      //   单患者复现 / 定向重摄受影响子集(分类修复后只重摄 reclassify 的患者)用。
      const onlyPatient = process.env.PAC_COHORT_ONLY_PATIENT?.trim();
      if (onlyPatient) {
        const ids = onlyPatient.split(',').map((s) => s.trim()).filter(Boolean);
        const quoted = ids.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
        whereParts.push(ids.length === 1 ? `${patient_key_column} = ${quoted}` : `${patient_key_column} IN (${quoted})`);
      }
      // --clinics=X,Y:把 cohort 收窄到「在这些诊所看过」的患者。诊所过滤挂在有 org 的事实表
      //   (cohort.clinic_scope),用与 cohort 相同的患者键求交集。⚠️ 只收窄「列哪些患者」,
      //   不碰任何资源查询 → 被选中患者的全部相关主体(跨所有诊所)照常摄入(不误删)。
      if (clinics && clinics.length > 0) {
        const cs = cohort.clinic_scope;
        if (!cs) {
          throw new Error(
            '--clinics 需要 manifest.sql_source.cohort.clinic_scope 配置(from_table / org_column)',
          );
        }
        const orgList = clinics.map((c) => `'${c.replace(/'/g, "''")}'`).join(', ');
        const keyCols = tenant_key_column
          ? `${patient_key_column}, ${tenant_key_column}`
          : patient_key_column;
        const inSubquery = `SELECT DISTINCT ${keyCols} FROM ${cs.from_table} WHERE ${cs.org_column} IN (${orgList})`;
        whereParts.push(
          tenant_key_column
            ? `(${keyCols}) IN (${inSubquery})`
            : `${patient_key_column} IN (${inSubquery})`,
        );
        this.logger.log(`[clickhouse·cohort] --clinics 收窄:${clinics.length} 家诊所 (${cs.from_table}.${cs.org_column})`);
      }
      // --since=<date> / --months=N:把 cohort 收窄到「该日期后有来诊」的患者,按 list_cursor_column
      //   (患者主档时间列 last_visit_time)。同样只收窄「列哪些患者」——被选中患者的全部历史
      //   (含更早的事件)照常摄入(不误删)。可与 --clinics 叠加(AND)。需 cohort.list_cursor_column。
      if (since) {
        if (!list_cursor_column) {
          throw new Error(
            '--since/--months 需要 manifest.sql_source.cohort.list_cursor_column 配置(患者主档时间列)',
          );
        }
        // DW 的 last_visit_time 是 String(ISO 文本):空串 < 任何日期,故 >= cutoff 天然排除空串;
        // ISO 日期字典序 = 时间序,直接串比较,不转类型(避 NO_COMMON_TYPE)。
        whereParts.push(`${list_cursor_column} >= '${since.replace(/'/g, "''")}'`);
        this.logger.log(`[clickhouse·cohort] --since 收窄:${list_cursor_column} >= ${since}`);
      }
      const whereSql = whereParts.length > 0 ? ` WHERE ${whereParts.join(' AND ')}` : '';
      // dev/ops:PAC_COHORT_LIMIT=N → 只取 N 个患者(本地抽样重摄用)。不设 = 全量,默认行为不变。
      //   PAC_COHORT_SAMPLE = recent(默认,最近就诊优先)| oldest(最久未来,lapsed,召回候选多)
      //                     | random(随机,production 代表性,自然带出 ~召回率比例)
      const cohortLimitRaw = parseInt(process.env.PAC_COHORT_LIMIT ?? '', 10);
      const cohortLimit = Number.isFinite(cohortLimitRaw) && cohortLimitRaw > 0 ? cohortLimitRaw : 0;
      const sampleMode = (process.env.PAC_COHORT_SAMPLE ?? 'recent').toLowerCase();
      let extraSelect = '';
      let orderTail = ` ORDER BY ${patient_key_column}`; // 全量:按 key 稳定排序
      if (cohortLimit > 0) {
        if (sampleMode === 'random') {
          extraSelect = ', rand() AS _samp'; // 放进 SELECT 规避 DISTINCT + ORDER BY 非选列限制
          orderTail = ` ORDER BY _samp LIMIT ${cohortLimit}`;
        } else if (list_cursor_column && !unionMode) {
          const dir = sampleMode === 'oldest' ? 'ASC' : 'DESC';
          extraSelect = `, ${list_cursor_column}`;
          orderTail = ` ORDER BY ${list_cursor_column} ${dir} LIMIT ${cohortLimit}`;
        } else {
          orderTail = ` ORDER BY ${patient_key_column} LIMIT ${cohortLimit}`;
        }
      }
      const selectCols =
        (tenant_key_column ? `${patient_key_column}, ${tenant_key_column}` : patient_key_column) + extraSelect;
      const listFrom = unionMode ? `(${incBranches.join(' UNION ALL ')})` : patient_list_from;
      const sql = `SELECT DISTINCT ${selectCols} FROM ${listFrom}${whereSql}${orderTail}`;
      this.logger.log(
        `[clickhouse·cohort] list patient keys${unionMode ? `(union of ${incBranches.length} cursored tables)` : ''} — ${sql.slice(0, 300)}`,
      );
      const started = Date.now();
      const rows = (await this.queryJsonWithRetry(client, sql, 'list-patient-keys')) as Array<Record<string, unknown>>;
      const keys: CohortKey[] = rows.map((r) => ({
        key: String(r[patient_key_column] ?? ''),
        tenant: tenant_key_column ? String(r[tenant_key_column] ?? '') : undefined,
      }));
      this.logger.log(
        `[clickhouse·cohort]   → ${keys.length} distinct cohort key${tenant_key_column ? ' (含 tenant)' : ''},${Date.now() - started} ms`,
      );
      return keys;
    } finally {
      await client.close();
    }
  }

  /**
   * 加载指定 cohort key 集合的所有 source 表(cohort batch 模式入口)。
   *
   * ⭐ 宿主无关:cohort 过滤列从 manifest.sql_source.cohort 读,不硬编码。
   *
   * 跟 loadAllTables 的差别:
   *   - 每张表 SQL 注入 cohort 过滤:
   *       有 tenant_key:`(patient_key, tenant_key) IN ((k1,t1),(k2,t2),...)`
   *       无 tenant_key:`patient_key IN (k1,k2,...)`
   *   - 增量 cursor 跟全量一样支持(二者 AND 并存)
   *   - 不做反向拉主档(本批 patient 主档已含本批所有 patient,无需补)
   */
  async loadTablesForCohort(
    source: ClickHouseSource,
    cohortKeys: ReadonlyArray<CohortKey>,
    incremental?: IncrementalConfig,
  ): Promise<Record<string, unknown[]>> {
    const cohort = source.cohort;
    if (!cohort) {
      throw new Error('loadTablesForCohort 需要 manifest.sql_source.cohort 配置');
    }
    const conn = this.resolveConnection(source);
    const client = createClient({
      url: conn.url,
      database: conn.database,
      username: conn.username,
      password: conn.password,
      request_timeout: 60_000,
      compression: { response: true, request: false },
    });

    // 构造 cohort IN 子句(按是否配 tenant_key 分两种形态)
    const cohortClause = this.buildCohortClause(cohort, cohortKeys);

    const tables: Record<string, unknown[]> = {};
    const defaultLimit = source.default_limit ?? 100_000;

    try {
      // ⭐ PR5a:N 个 query 并行(远程 DW 往返是瓶颈,串行 N×170ms → 并行 ~170ms)
      //   @clickhouse/client 基于 HTTP,单 client 可并发多请求;每个 query 独立 result
      // 先初始化 cursorAdvances(避免并行 callback 各自 ?? {} 互相覆盖的竞态)
      if (incremental) incremental.cursorAdvances = incremental.cursorAdvances ?? {};
      const queryStart = Date.now();
      await Promise.all(
        Object.entries(source.queries).map(async ([tableName, sql]) => {
          const incCfg = incremental?.perQuery[tableName];
          const sqlWithCursor = incCfg
            ? this.injectIncrementalCursor(sql, incCfg.cursorColumn, incCfg.cursorValue)
            : sql;
          const sqlWithCohort = this.injectCohortFilter(sqlWithCursor, cohortClause);
          const sqlWithLimit = this.applyDefaultLimit(sqlWithCohort, defaultLimit);
          const started = Date.now();
          const rows = await this.queryJsonWithRetry(client, sqlWithLimit, `cohort:${tableName}`);
          tables[tableName] = rows;
          this.logger.log(
            `[clickhouse·cohort]   "${tableName}" → ${rows.length} 行,${Date.now() - started} ms`,
          );
          // cursor advance(并发安全:cursorAdvances 已预初始化,每个 tableName 写自己的 key)
          if (incremental?.cursorAdvances && incCfg && rows.length > 0) {
            const maxVal = this.computeMax(rows as Record<string, unknown>[], incCfg.cursorColumn);
            if (maxVal) {
              const prev = incremental.cursorAdvances[tableName];
              if (!prev || maxVal > prev) {
                incremental.cursorAdvances[tableName] = maxVal;
              }
            }
          }
        }),
      );
      this.logger.debug(`[clickhouse·cohort] ${Object.keys(source.queries).length} query 并行完成 ${Date.now() - queryStart}ms`);
    } finally {
      await client.close();
    }
    return tables;
  }

  /// 按 cohort 配置构造 IN 子句(有 tenant_key → tuple 形态;无 → 单列形态)
  private buildCohortClause(
    cohort: NonNullable<ClickHouseSource['cohort']>,
    keys: ReadonlyArray<CohortKey>,
  ): string {
    const esc = (s: string) => s.replace(/'/g, "''");
    if (cohort.tenant_key_column) {
      const tuples = keys
        .map((k) => `('${esc(k.key)}', '${esc(k.tenant ?? '')}')`)
        .join(',');
      return `(${cohort.patient_key_column}, ${cohort.tenant_key_column}) IN (${tuples})`;
    }
    const list = keys.map((k) => `'${esc(k.key)}'`).join(',');
    return `${cohort.patient_key_column} IN (${list})`;
  }

  /// 把 cohort IN 子句注入到已有 SQL 的 WHERE 末尾(无 WHERE 时新建)
  private injectCohortFilter(sql: string, cohortClause: string): string {
    if (!cohortClause) return sql; // 空 cohort 不注入(理论上不该发生)
    // 找最外层 WHERE(若有);若已有 ORDER BY/LIMIT/GROUP BY,要插在它们之前
    const hasWhere = /\bWHERE\b/i.test(sql);
    if (!hasWhere) {
      // 在 ORDER/LIMIT/GROUP 之前 / FROM 之后插 WHERE
      const tailMatch = sql.match(/(\s+(?:ORDER\s+BY|LIMIT|GROUP\s+BY)\b[\s\S]*)$/i);
      if (tailMatch) {
        const tail = tailMatch[1]!;
        return sql.slice(0, sql.length - tail.length) + ` WHERE ${cohortClause}` + tail;
      }
      return `${sql} WHERE ${cohortClause}`;
    }
    // 已有 WHERE:在 ORDER/LIMIT/GROUP 之前 / WHERE body 末尾插 AND ...
    const tailMatch = sql.match(/(\s+(?:ORDER\s+BY|LIMIT|GROUP\s+BY)\b[\s\S]*)$/i);
    if (tailMatch) {
      const tail = tailMatch[1]!;
      return sql.slice(0, sql.length - tail.length) + ` AND ${cohortClause}` + tail;
    }
    return `${sql} AND ${cohortClause}`;
  }

  /// 把原 SQL 的 cohort/cursor/ORDER/LIMIT 全部剥离,改写为 patient_id+brand 精确过滤
  private injectPatientFilter(
    originalSql: string,
    patientIdEsc: string,
    brandEsc: string,
  ): string {
    // 包裹原查询为子查询,外层按 patient_id+brand 过滤 —— 兼容任意复杂 SQL。
    // ⚠️ 不再用 regex 解析 `SELECT cols FROM table`:遇到嵌套子查询(如 image_finding 的
    //   `SELECT … FROM (…s1… ARRAY JOIN …) s2 WHERE tooth != ''`)会把半截子查询拖进列表 →
    //   括号不匹配 → CH 语法错误(单患者刷新 500)。包裹法对简单/嵌套查询都成立:
    //   所有 cohort 资源查询的最外层都暴露 patient_id + brand;原 cohort IN 子查询与业务过滤一并保留。
    const inner = originalSql.trim().replace(/;\s*$/, '');
    return `SELECT * FROM (\n${inner}\n) AS _pt WHERE patient_id = '${patientIdEsc}' AND brand = '${brandEsc}'`;
  }

  /// W4 末:重写 SQL,把 cursor 条件注入到原 SQL 的 cohort 子查询和外层 WHERE
  /// 策略:增量模式去掉 dev cohort LIMIT,只按 cursor 过滤
  ///   - 找最外层 FROM <table>,在它之后 inject WHERE cursor_column > 'cursor_value'
  ///   - 同时去掉 cohort 子查询(WHERE (patient_id, brand) IN (...))跟 ORDER BY/LIMIT
  /// 简化版:直接重写为 `SELECT <orig_columns> FROM dw_group.<table> WHERE cursor_column > '...'`
  /// 假设:cold-import 的 query 已 SELECT 出 cursor_column(本任务 manifest 已加)
  private injectIncrementalCursor(
    originalSql: string,
    cursorColumn: string,
    cursorValue: string | null,
  ): string {
    // 提取 SELECT 子句 (FROM 之前) 跟 FROM <table>
    const m = originalSql.match(/^\s*SELECT\s+([\s\S]+?)\s+FROM\s+([\w.]+)/i);
    if (!m) {
      throw new Error(
        `[incremental] cannot parse SQL for cursor injection: ${originalSql.slice(0, 80)}...`,
      );
    }
    const selectCols = m[1]!.trim();
    const fromTable = m[2]!;
    // 收集所有 WHERE 子句(cursor + 业务过滤),最后统一 'WHERE ... AND ...' 拼装
    const clauses: string[] = [];
    if (cursorValue) {
      clauses.push(`${cursorColumn} > '${cursorValue.replace(/'/g, "''")}'`);
    }
    // 业务过滤(如 settlement_status=1)— 去掉 cohort 子查询(patient_id,brand IN ...)
    clauses.push(...this.extractBusinessFilters(originalSql));
    const whereSql = clauses.length > 0 ? ` WHERE ${clauses.join(' AND ')}` : '';
    return `SELECT ${selectCols} FROM ${fromTable}${whereSql} ORDER BY ${cursorColumn}`;
  }

  /// 从原 SQL 抽出 WHERE 子句里**非 cohort** 的业务过滤条件
  /// 返回 clause 数组(每个 clause 不带 WHERE/AND 前缀,纯条件)
  private extractBusinessFilters(sql: string): string[] {
    // 抓 WHERE ... 到 ORDER BY/LIMIT/end 之间
    const m = sql.match(/\bWHERE\s+([\s\S]+?)(?:\bORDER\s+BY\b|\bLIMIT\b|$)/i);
    if (!m) return [];
    const whereBody = m[1]!.trim();
    // 拆 AND,去掉:① cohort 子查询 ② 老的 last_visit_time IS NOT NULL(增量不需要)
    return this.splitAndClauses(whereBody)
      .map((c) => c.trim())
      .filter(
        (c) =>
          c &&
          !/\(patient_id\s*,\s*brand\)\s+IN/i.test(c) &&
          !/^last_visit_time\s+IS\s+NOT\s+NULL$/i.test(c),
      );
  }

  private splitAndClauses(whereBody: string): string[] {
    // 按 ` AND `(大小写)拆分,但跳过括号内的 AND
    const out: string[] = [];
    let depth = 0;
    let buf = '';
    const tokens = whereBody.split(/(\bAND\b)/i);
    for (const tok of tokens) {
      for (const ch of tok) {
        if (ch === '(') depth++;
        else if (ch === ')') depth--;
      }
      if (/^\s*AND\s*$/i.test(tok) && depth === 0) {
        out.push(buf);
        buf = '';
      } else {
        buf += tok;
      }
    }
    if (buf.trim()) out.push(buf);
    return out;
  }

  /// 算行集合中 cursor_column 的字符串 max(host updated_date 是 String 类型)
  private computeMax(rows: Record<string, unknown>[], column: string): string | null {
    let max: string | null = null;
    for (const r of rows) {
      const v = r[column];
      if (v === null || v === undefined) continue;
      const s = String(v);
      if (!max || s > max) max = s;
    }
    return max;
  }

  /**
   * 给没显式带 LIMIT 的 SQL 自动追加 LIMIT,防内存炸。
   * 简单字符串嗅探,匹配 LIMIT N(忽略大小写,允许末尾分号)。
   */
  private applyDefaultLimit(sql: string, defaultLimit: number): string {
    const trimmed = sql.trim().replace(/;\s*$/, '');
    // 已有 LIMIT N / LIMIT N, M / LIMIT N OFFSET M 在末尾 → 不再追加
    if (/\bLIMIT\s+\d+(\s*,\s*\d+|\s+OFFSET\s+\d+)?\s*$/i.test(trimmed)) return trimmed;
    return `${trimmed} LIMIT ${defaultLimit}`;
  }
}
