import { z } from 'zod';
import { TransformsSchema } from '../transforms/transforms.schema';

/**
 * Cold Import Manifest(v2.2 — 双源 + Layer A.5 transforms 单 yaml 方案)
 *
 * 宿主导出脱敏数据时附带的 `manifest.yaml`。host 只暴露**原始表**,
 * 由 PAC 这边的 yaml(transforms + assemblers)做形态改造 + 翻译。
 *
 * v2.2 新增:
 *   - 顶层 `transforms[]` — Layer A.5 声明式形态改造(6 operator 白名单)
 *     - SQL 退化为"等价 csv 单表 dump 的 SELECT"(只列选 + WHERE 过滤)
 *     - JSON 数组拆行 / 字段派生 / 多列推断 / 关键词分流 全在 transforms
 *     - 实现双源真等价(直连 ClickHouse 或 csv 文件,下游无感知)
 *
 * v2.1 ClickHouse 直连模式:跳过 dump 文件中间步,PAC 直接 SQL 拉到内存。
 *
 * 详见 apps/pac-docs/content/docs/architecture/data-ingestion.mdx §五(Layer A.5)。
 */

/// 原始表声明 — 文件 + 表名(供 assembler 索引)
export const RawTableSchema = z.object({
  /// host 表名(被 assembler.primary.table / join_arrays[*].table 引用)
  table: z.string().min(1),
  /// 文件相对路径(相对于 manifest 所在目录)
  file: z.string().min(1).describe('e.g. tb_encounter.csv / tb_patient.json'),
  /// 文件格式(默认按扩展名推断)
  format: z.enum(['csv', 'json']).optional(),
  /// 成组主体标签(宿主特有,可选)。同 group 的表在**网页上传**时要么全在、要么整组跳过
  ///   —— 由 ImportUploadService 通用地读此声明做完整性校验。用于"半组产不出主体"的强依赖
  ///   (如 treatment_planned = customer_treat_plan[头] + _item[行],缺任一整个资源产 0 行)。
  ///   不标 = 独立表:缺失仅少那部分 facts(装载层已宽容跳过)。
  ///   注:CLI 全量导入不受此约束(只 warn 跳过);仅上传路径按 group 前置守卫。
  group: z.string().min(1).optional(),
});
export type RawTable = z.infer<typeof RawTableSchema>;

/// Assembler 配置文件引用
export const AssemblerRefSchema = z.object({
  /// assembler yaml 相对路径(相对于 manifest 所在目录)
  file: z.string().min(1).describe('e.g. assemblers/clinical_encounter.yaml'),
});
export type AssemblerRef = z.infer<typeof AssemblerRefSchema>;

/**
 * ClickHouse SQL 数据源(替代文件 tables[])
 *
 * 短期 cold-import 直连模式 — 跳过"DW dump → CSV 落盘 → PAC 读"中间步,
 * PAC 直接 SQL 拉到内存当 tables 喂 AssemblerEngine。
 *
 * 适用:DW 已 ETL 到 ClickHouse,PAC 一次性灌历史。
 * 后续:W4+ 升级为带 cursor 的真增量 pull。
 *
 * 密码:从环境变量读取(`password_env`),不允许 plaintext 进 yaml。
 */
/// 增量水位声明:表名 → cursor 列。**与源类型解耦**(2026-07 统一):
///   - sql_source 内声明(jvs-dw 现状,位置不动):拉模式,读水位注 WHERE + 写水位
///   - manifest 顶层声明(file 源用):file 装载不消费水位(全量装载,幂等去重兜底),
///     只做**记账**——同一段 finally 写 cursor_after = run_start,水位供 delta 导出
///     WHERE 模板 / push 回放起点使用。两处都写时 sql_source 内优先。
export const IncrementalDeclSchema = z.object({
  /// per query/table 配置;表名 → cursor_column
  /// 例:fact_emr_treatment_out: updated_date / med_emr_info: updated_gmt_at
  per_query: z.record(
    z.string(),
    z.object({
      cursor_column: z.string().min(1),
    }),
  ),
});

export const ClickHouseSourceSchema = z.object({
  kind: z.literal('clickhouse'),
  connection: z.object({
    /// ClickHouse HTTP/HTTPS URL,e.g. https://cc-xxx.clickhouse.aliyuncs.com:8443
    url: z.string().url(),
    database: z.string().min(1),
    username: z.string().min(1),
    /// 环境变量名;PAC 启动期从 process.env 读
    password_env: z.string().min(1),
  }),
  /// 资源映射:tableName → SQL
  /// 查到的行会以该 tableName 作为 key 喂 AssemblerEngine(对应 assembler.primary.table)
  /// SQL 一般直接 SELECT * FROM <表>,需要 WHERE 过滤试点诊所 / 时间窗
  queries: z.record(z.string(), z.string().min(1)),
  /// 每个 query 上限(防内存炸);默认 100k 行
  default_limit: z.number().int().positive().default(100000).optional(),
  /// W4 末:DW 增量配置(per table cursor column)
  /// 跑增量模式时 PAC 会读 sync_logs 上次 cursor_after,注入 WHERE cursor_column > '...'
  /// 跑完写新 cursor = run_start 到 sync_logs
  /// 首次跑(无 cursor)= 全量,后续 = 增量
  incremental: IncrementalDeclSchema.optional(),
  /// ⭐ Cohort 分批配置(PR2/PR4)— 宿主无关性的关键:
  ///   patient 列表来源 + 主键列 + 租户区分列全部声明在这,代码不硬编码任何表名/列名。
  ///   不配 cohort → 不分批(single-shot,文件源 / 小数据集)。
  ///   各 host 只改本段 + queries SQL,摄入流程代码零改动。
  cohort: z
    .object({
      /// 列患者清单的源表全名(库.表);通常是 patient 主档表
      /// 例:dw_group.fact_client_out
      patient_list_from: z.string().min(1),
      /// 患者主键列名 — 所有源表共用此列名做 (key) IN (...) cohort 过滤
      /// 例:patient_id
      patient_key_column: z.string().min(1).default('patient_id'),
      /// 租户区分列名(可选)— 多 tenant per host 时用(jvs-dw 用 brand 区分瑞尔/瑞泰)
      /// 配了 → cohort 用 (patient_key, tenant_key) 复合键;不配 → 仅 patient_key 单键
      tenant_key_column: z.string().min(1).optional(),
      /// 列患者时的增量 cursor 列(对应 patient_list_from 表的时间列)
      /// 例:last_visit_time;不配 → 列患者不做增量过滤(总是全量列再 cohort 切)
      list_cursor_column: z.string().min(1).optional(),
      /// 可选:诊所范围过滤源 —— 供 cold-import 的 `--clinics=X,Y` 把 cohort 收窄到
      ///   「在这些诊所看过的患者」。patient 主档(patient_list_from)通常无 org 列,
      ///   故诊所过滤挂在有 org 的事实表上。例:
      ///     { from_table: dw_group.fact_emr_treatment_out, org_column: organization_id }
      ///   ⚠️ 被动声明:不传 `--clinics` 时完全不生效(默认仍全量)。只收窄「列哪些患者」,
      ///      不碰任何资源查询 → 被选中患者的全部相关主体(跨所有诊所)照常摄入。
      clinic_scope: z
        .object({
          from_table: z.string().min(1),
          org_column: z.string().min(1).default('organization_id'),
        })
        .optional(),
      /// 增量「反向拉主档」的来源表(queries 里的 key,不带库前缀)。
      ///
      /// 【解决什么】主档 cursor 是 list_cursor_column(jvs-dw = last_visit_time),
      ///   患者不来诊则主档拉不到他 —— 但他的**事实**可能变了(EMR 被编辑、复杂病例被关闭)。
      ///   增量跑完后从这些表里收集患者键,反向补拉一次主档,保证"事实变了主档也在场"。
      ///
      /// 【为什么要可配】此前是四张 jvs-dw 表名硬编码在 service 里,新增一张要改通用代码;
      ///   而"哪些表的变化该带出主档"本就是**宿主口径**(取决于该宿主哪些表有独立 cursor)。
      ///   不配 → 保持历史默认(见 DEFAULT_REVERSE_PULL_FROM),行为不变。
      ///
      /// ⚠️ 表的行必须带 patient_key_column / tenant_key_column 两列(SELECT 里 AS 别名亦可),
      ///   否则收集不到键、该表静默不参与反向拉。
      reverse_pull_from: z.array(z.string().min(1)).optional(),
    })
    .optional(),
});

/// 反向拉主档的历史默认表集(jvs-dw 首版硬编码,保留为默认值以免既有 manifest 行为变化)。
/// 新宿主 / 新增表请在 manifest 的 cohort.reverse_pull_from 显式声明。
export const DEFAULT_REVERSE_PULL_FROM = [
  'fact_appointment_out',
  'fact_emr_treatment_out',
  'fact_settlement_out',
  'fact_settlement_mode_out',
] as const;
export type ClickHouseSource = z.infer<typeof ClickHouseSourceSchema>;

export const ColdImportManifestSchema = z
  .object({
    /// 关联到 PAC hosts.name(slug,平台必须先 seed/admin 注册)
    host_name: z.string().min(1),
    display_name: z.string().optional(),

    /// 是否纳入每日自动增量同步(SyncIncrementalSchedulerService 启动时自动发现)。
    /// 默认 false:host 接入(冷启)后,显式置 true 才会被每晚 cron 自动拉。
    /// 取代旧 `PAC_INCREMENTAL_HOSTS` env 清单 —— 声明式、跟 host 配置同处、不会 drift
    ///(接入新 host 忘了改 env 导致静默不同步的坑)。env 仍可作显式 override(见 scheduler)。
    auto_sync: z.boolean().optional().default(false),

    /// **每宿主增量 cron**(可选)—— 上游 DW 刷新窗口因 host 而异,该 host 在自己 DW 就绪后跑。
    /// 缺省回退全局 `PAC_INCREMENTAL_CRON`。cron 按本 manifest 的 `timezone` 解释(不是服务器 UTC)。
    /// 例:'30 2 * * *'(DW 02:00 刷新后)。仅 auto_sync=true 时生效。
    incremental_cron: z.string().min(1).optional(),

    /// **每宿主监控阈值**(可选)—— 不同 host 推送/刷新节奏不同,一刀切阈值要么误报要么漏报。
    /// 缺省回退全局 env(见 dw-lag-monitor)。
    monitoring: z
      .object({
        /// DW 增量游标滞后告警(pull host):warn / error 小时阈值
        dw_lag_warn_hours: z.number().positive().optional(),
        dw_lag_error_hours: z.number().positive().optional(),
        /// push 断流告警(push host):超过此小时数无 push → critical。
        /// 实时推的 host 设小(如 3),每晚批量推的设大(如 26)。
        push_lag_error_hours: z.number().positive().optional(),
      })
      .optional(),

    /// **单 tenant 场景**:静态 tenant_id,所有行共用一个 tenant
    tenant_id: z.string().min(1).optional(),

    /// **多 tenant 场景**(v2.3,W3 末实装):
    /// 单 host 内部多个 tenant(如 jvs-dw 含瑞尔/瑞泰双品牌),
    /// PAC 按行内 `tenant_field` 字段值直接当 tenant_id。
    ///
    /// 例:tenant_field='brand',row.brand='瑞尔' → tenant_id='瑞尔';row.brand='瑞泰' → '瑞泰'。
    ///
    /// **不预设值集** — PAC 不知道也不需要知道 host 会给哪些 brand,数据来什么就是什么。
    /// 这跟 "host 给的数据范围就是 PAC 该消化的范围" 一致(过滤归 host 端责任)。
    tenant_field: z.string().min(1).optional(),

    /// **DEPRECATED**(v2.3 起 tenant_field 直接用字段值;保留兼容老 yaml 显式映射)
    /// 若同时给 tenant_field + tenant_map,先查 map(`map[row[field]] ?? row[field]`)。
    tenant_map: z.record(z.string(), z.string().min(1)).optional(),
    brand_to_tenant: z.record(z.string(), z.string()).optional(),

    /// **患者命名空间字段**(集团模型):tenant=集团后,external_id 在集团内可撞号
    /// (不同源组织发的同号 = 不同人)。声明用哪个**源行字段**填 patients.source_unit 消歧。
    /// 通用 — 各 host "在哪层发患者号"不同:jvs-dw='brand'(瑞尔/瑞泰);别的 host 可能=区域/诊所/不设。
    /// 不设时 source_unit='',适合单命名空间 host。
    identity_namespace_field: z.string().min(1).optional(),

    /// 金额单位(yuan = 元小数,fen = 分整数)
    amount_unit: z.enum(['fen', 'yuan']),
    /// 时区(用于解析无 tz 的 datetime 字符串)
    timezone: z.string().min(1).describe('IANA, e.g. Asia/Shanghai'),

    /// 文件数据源(跟 sql_source 二选一)
    tables: z.array(RawTableSchema).optional(),

    /// ClickHouse SQL 数据源(跟 tables 二选一)
    sql_source: ClickHouseSourceSchema.optional(),

    /// 顶层增量水位声明 —— **file 源专用**(sql_source 宿主继续在 sql_source.incremental 声明,
    /// 两处都写时 sql_source 内优先)。file 装载不消费水位,仅记账 cursor_after = run_start
    /// (同 sql_source 宿主同一段 finally,逻辑一致);详见 IncrementalDeclSchema 注释。
    incremental: IncrementalDeclSchema.optional(),

    /// 诊所名字典来源(可选)—— 声明哪张表的哪两列 = 诊所 id → 名字。
    /// `refresh-clinic-names` CLI 据此 SELECT DISTINCT 出 id→名,写 host.clinicNames;
    /// /auth/session 再合并进 dictionary.clinics 下发,前端显示名字不依赖登录传。
    /// sql_source 模式:table 是 CH 表名(或 queries 的 key);文件模式:table 对应 tables[].table。
    clinic_directory: z
      .object({
        table: z.string().min(1),
        id_field: z.string().min(1),
        name_field: z.string().min(1),
      })
      .optional(),

    /// **Layer A.5 transforms**(可选)— 在 sources 之后、AssemblerEngine 之前跑。
    /// 6 operator 白名单:project / split_json_array / derive / route_by_pattern / pick_first_nonzero / filter。
    /// 详见 apps/pac-docs/content/docs/architecture/data-ingestion.mdx §五。
    transforms: TransformsSchema.optional(),

    /// PAC 写的 assembler 配置清单(每个 canonical resource 一份)
    assemblers: z.array(AssemblerRefSchema).min(1),

    notes: z.string().optional(),
  })
  .refine(
    (m) => !!m.tables || !!m.sql_source,
    'manifest 必须含 tables[](文件源)或 sql_source(ClickHouse 直连)其中之一',
  )
  .refine(
    (m) => !!m.tenant_id || !!m.tenant_field || !!m.brand_to_tenant,
    'manifest 必须含 tenant_id(单 tenant 静态)/ tenant_field(per-row 取字段值)/ brand_to_tenant(兼容) 之一',
  )
  .refine(
    (m) => !m.tenant_map || !!m.tenant_field,
    '配置 tenant_map 时必须同时给 tenant_field(从哪个字段读 tenant 标识)',
  );
export type ColdImportManifest = z.infer<typeof ColdImportManifestSchema>;

export function inferFileFormat(file: string): 'csv' | 'json' {
  if (file.toLowerCase().endsWith('.json')) return 'json';
  if (file.toLowerCase().endsWith('.csv')) return 'csv';
  throw new Error(`Cannot infer format from extension: ${file}`);
}
