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

/// 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);

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

  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 {
      for (const [tableName, sql] of Object.entries(source.queries)) {
        // 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);
        this.logger.log(
          `[clickhouse] query "${tableName}" — ${sqlWithLimit.slice(0, 120).replace(/\s+/g, ' ')}...`,
        );
        const started = Date.now();
        const result = await client.query({
          query: sqlWithLimit,
          format: 'JSONEachRow',
        });
        const rows = (await result.json()) as unknown[];
        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 && incCfg && rows.length > 0) {
          const maxVal = this.computeMax(rows as Record<string, unknown>[], incCfg.cursorColumn);
          if (maxVal) {
            incremental.cursorAdvances = incremental.cursorAdvances ?? {};
            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 {
      for (const [tableName, sql] of Object.entries(source.queries)) {
        const patientSql = this.injectPatientFilter(sql, pidEsc, brandEsc);
        this.logger.log(
          `[clickhouse·patient] query "${tableName}" — ${patientSql.slice(0, 140).replace(/\s+/g, ' ')}...`,
        );
        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 (patient_id, brand) pair,作为 cohort batch 的 ORDER 边界。
   *
   * 用 fact_client_out 当 source of truth(它就是 patient 主档表);
   * 增量模式 cursorAfter 用 last_visit_time 列(同 manifest.sql_source.incremental.per_query.fact_client_out)。
   *
   * ORDER BY patient_id ASC 保证 OFFSET LIMIT 分页稳定;
   * 跨 batch 时同一 patient 不会拆开(LIMIT 5000 = 拿 5000 个 distinct patient)。
   */
  async listPatientPairs(
    source: ClickHouseSource,
    incremental?: IncrementalConfig,
  ): Promise<Array<{ patient_id: string; brand: string }>> {
    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 {
      const cursorCfg = incremental?.perQuery['fact_client_out'];
      const whereParts: string[] = [];
      if (cursorCfg?.cursorValue) {
        whereParts.push(
          `${cursorCfg.cursorColumn} > '${cursorCfg.cursorValue.replace(/'/g, "''")}'`,
        );
      }
      const whereSql = whereParts.length > 0 ? ` WHERE ${whereParts.join(' AND ')}` : '';
      const sql = `SELECT DISTINCT patient_id, brand FROM dw_group.fact_client_out${whereSql} ORDER BY patient_id`;
      this.logger.log(`[clickhouse·cohort] list patient pairs — ${sql.slice(0, 200)}`);
      const started = Date.now();
      const result = await client.query({ query: sql, format: 'JSONEachRow' });
      const rows = (await result.json()) as Array<{ patient_id: string; brand: string }>;
      this.logger.log(
        `[clickhouse·cohort]   → ${rows.length} distinct (patient_id, brand) pairs, ${Date.now() - started} ms`,
      );
      return rows;
    } finally {
      await client.close();
    }
  }

  /**
   * 加载指定 patient 集合的所有 source 表(cohort batch 模式入口)。
   *
   * 跟 loadAllTables 的差别:
   *   - 每张表 SQL 注入 `(patient_id, brand) IN ((p1,b1),(p2,b2),...)` 过滤
   *   - 增量 cursor 跟全量一样支持(WHERE cursor_col > val AND in (...) 二者并存)
   *   - 不做反向拉主档(本批 fact_client_out 已含本批所有 patient 主档,无需补)
   *   - default_limit 仍然应用(防御性,本批数据通常远小于 limit)
   *
   * 用途:importDirectory 在 cohort batch 模式下按 patient 段调用,
   * 每批跑完释放内存,bound 资源使用。
   */
  async loadTablesForCohort(
    source: ClickHouseSource,
    pairs: ReadonlyArray<{ patient_id: string; brand: string }>,
    incremental?: IncrementalConfig,
  ): Promise<Record<string, unknown[]>> {
    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 },
    });

    // 构造 IN tuple 列表(同 reversePullPatientMaster 风格)
    const tuples = pairs
      .map(
        (p) =>
          `('${(p.patient_id ?? '').replace(/'/g, "''")}', '${(p.brand ?? '').replace(/'/g, "''")}')`,
      )
      .join(',');

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

    try {
      for (const [tableName, sql] of Object.entries(source.queries)) {
        const incCfg = incremental?.perQuery[tableName];
        // 1. cursor 注入(同 loadAllTables 路径)
        const sqlWithCursor = incCfg
          ? this.injectIncrementalCursor(sql, incCfg.cursorColumn, incCfg.cursorValue)
          : sql;
        // 2. cohort 注入:在 WHERE 后追加 AND (patient_id, brand) IN (...)
        const sqlWithCohort = this.injectCohortFilter(sqlWithCursor, tuples);
        const sqlWithLimit = this.applyDefaultLimit(sqlWithCohort, defaultLimit);
        this.logger.log(
          `[clickhouse·cohort] query "${tableName}" — ${sqlWithLimit.slice(0, 140).replace(/\s+/g, ' ')}...`,
        );
        const started = Date.now();
        const result = await client.query({ query: sqlWithLimit, format: 'JSONEachRow' });
        const rows = (await result.json()) as unknown[];
        tables[tableName] = rows;
        this.logger.log(
          `[clickhouse·cohort]   "${tableName}" → ${rows.length} 行,${Date.now() - started} ms`,
        );
        // 增量 cursor 推进(同 loadAllTables;cohort 模式下也累积,run_start cursor 会覆盖)
        if (incremental && incCfg && rows.length > 0) {
          const maxVal = this.computeMax(rows as Record<string, unknown>[], incCfg.cursorColumn);
          if (maxVal) {
            incremental.cursorAdvances = incremental.cursorAdvances ?? {};
            const prev = incremental.cursorAdvances[tableName];
            if (!prev || maxVal > prev) {
              incremental.cursorAdvances[tableName] = maxVal;
            }
          }
        }
      }
    } finally {
      await client.close();
    }
    return tables;
  }

  /// 把 (patient_id, brand) IN tuple 过滤注入到已有 SQL 的 WHERE 末尾(无 WHERE 时新建)
  private injectCohortFilter(sql: string, tuplesCsv: string): string {
    if (!tuplesCsv) return sql; // 空 cohort 不注入(理论上不该发生)
    const cohortClause = `(patient_id, brand) IN (${tuplesCsv})`;
    // 找最外层 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 {
    const m = originalSql.match(/^\s*SELECT\s+([\s\S]+?)\s+FROM\s+([\w.]+)/i);
    if (!m) {
      throw new Error(
        `[patient-refresh] cannot parse SQL: ${originalSql.slice(0, 80)}...`,
      );
    }
    const selectCols = m[1]!.trim();
    const fromTable = m[2]!;
    const clauses: string[] = [
      `patient_id = '${patientIdEsc}'`,
      `brand = '${brandEsc}'`,
    ];
    // 业务过滤(如 settlement_status=1 / is_refund=0 / appo_status IN ... )保留
    clauses.push(...this.extractBusinessFilters(originalSql));
    return `SELECT ${selectCols} FROM ${fromTable} WHERE ${clauses.join(' AND ')}`;
  }

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