/**
 * Transform Pipeline Schema(Layer A.5)
 *
 * 定位:在 sources(SQL/CSV)和 AssemblerEngine 之间插一层 yaml 声明式形态改造。
 *
 * 纪律(canonical-fact-layer.md 原则 1):
 *   - 5 个 operator 白名单,**不允许任意 JS 表达式 / eval / 字符串模板填代码**
 *   - 声明式、无副作用、纯函数变换
 *   - 撞墙必须加新 operator(走 PR + review),不允许 hack
 *
 * 6 operator(对应 SQL 干过的事):
 *   ① project           — 选列 / 改名(等价 SELECT col1, col2 [AS alias])
 *   ② split_json_array  — JSON 数组字段拆行(等价 ARRAY JOIN + JSONExtract)
 *   ③ derive            — 字段派生(substring / concat / lower / upper / default)
 *   ④ route_by_pattern  — 按字段值分流到多个 output 表
 *   ⑤ pick_first_nonzero — 多个数值列挑第一个非零对应 label(等价 multiIf pay_* > 0)
 *   ⑥ filter            — 多字段行过滤(等价 SQL WHERE 行级过滤)
 *
 * 数据流:
 *   sources (raw tables) ──→ transforms[] ──→ tables dictionary ──→ AssemblerEngine
 *
 * 双源等价:transforms 输入是 row 数组,跟 SQL 直连或 CSV 文件无关。
 */
import { z } from 'zod';

// ─────────────────────────────────────────────────────────
// 通用类型
// ─────────────────────────────────────────────────────────

const tableNameSchema = z
  .string()
  .min(1)
  .describe('上下游表名(可以引用 source table 或前置 transform output)');

// ─────────────────────────────────────────────────────────
// ① project — 选列 / 改名
// ─────────────────────────────────────────────────────────

/**
 * 等价 SQL `SELECT col1, col2 AS alias FROM input`。
 * - keep:保留哪些列(白名单)。不写 = 全保留。
 * - rename:列改名(source → target);跟 keep 联动 — keep 数组写 source 名,rename 在 keep 通过后做改名。
 */
export const ProjectOpSchema = z.object({
  kind: z.literal('project'),
  input: tableNameSchema,
  output: tableNameSchema,
  keep: z.array(z.string().min(1)).optional(),
  rename: z.record(z.string(), z.string()).optional(),
});
export type ProjectOp = z.infer<typeof ProjectOpSchema>;

// ─────────────────────────────────────────────────────────
// ② split_json_array — JSON 数组字段拆行
// ─────────────────────────────────────────────────────────

/**
 * 把主表某 JSON 数组字段拆成 N 行子记录。等价 ClickHouse ARRAY JOIN + JSONExtract。
 *
 * 输入:1 行 = 主表行,含一个 JSON 数组字段(stringified 或已 parse 的数组)
 * 输出:N 行 = 数组每元素一行,继承 parent_keys 字段,加上 element_fields 提取的字段
 *
 * parent_keys 形态:`{ outputName: parentField }` 例:`{ emr_id: id }` 表示
 *   父行 `id` 字段值在拆出来的子行里命名为 `emr_id`(防跟元素 id 字段冲突)
 *
 * element_fields 形态同上:`{ outputName: jsonPath }`,jsonPath 当前只支持单层(对象属性名)
 *
 * where 形态:对**拆出来的子行**做过滤,key=子行字段名,value=过滤规则。
 *   - `not_empty: true` — 字符串非空 / null 过滤
 *   - `in: [...]` — 值在列表中
 *   - `equals: x` — 值等于 x
 */
export const WhereRuleSchema = z.union([
  z.object({ not_empty: z.literal(true) }),
  z.object({ in: z.array(z.union([z.string(), z.number()])).min(1) }),
  z.object({ equals: z.union([z.string(), z.number()]) }),
]);
export type WhereRule = z.infer<typeof WhereRuleSchema>;

export const SplitJsonArrayOpSchema = z.object({
  kind: z.literal('split_json_array'),
  input: tableNameSchema,
  output: tableNameSchema,
  array_field: z.string().min(1).describe('主表里的 JSON 数组字段名'),
  parent_keys: z
    .record(z.string(), z.string())
    .default({})
    .describe('父表字段继承,outputName: parentField'),
  element_fields: z
    .record(z.string(), z.string())
    .describe('数组元素字段提取,outputName: jsonPath(单层属性名)'),
  where: z.record(z.string(), WhereRuleSchema).optional(),
});
export type SplitJsonArrayOp = z.infer<typeof SplitJsonArrayOpSchema>;

// ─────────────────────────────────────────────────────────
// ③ derive — 字段派生(限定纯函数 op)
// ─────────────────────────────────────────────────────────

/**
 * 给行加新字段(in-place,output 可以同 input)。
 *
 * 8 个内置 op:
 *   - substring:  { op: substring, from: field, start: 0, end: 3 }
 *   - concat:     { op: concat, parts: ["${field}" | "literal", ...] }
 *   - lower:      { op: lower, from: field }
 *   - upper:      { op: upper, from: field }
 *   - default:    { op: default, from: field, value: "fallback" } — 原值空时填 value
 *   - trim:       { op: trim, from: field } — 去首尾空白 + 尾部标点(。;,?!、\r\n),脏字段归一
 *   - normalize:  { op: normalize, from: field } — trim 升级版 + CJK→ASCII 中段标点
 *                  (中括号()/中逗号,/中分号;/中尖括号 < >)统一 ASCII,中文宿主 enum 匹配必备
 *   - coalesce:   { op: coalesce, from: [f1, f2, ...] } — 取第一个非空字段值(诊断码兜底:stdCode→message)
 *
 * concat parts 里 `${fieldName}` 是字段引用,其他视为字面量。
 * 限定不允许任意表达式(canonical-fact-layer.md 原则 1)。
 */
export const DeriveSubstringSchema = z.object({
  op: z.literal('substring'),
  from: z.string().min(1),
  start: z.number().int().nonnegative().default(0),
  end: z.number().int().optional(),
});

export const DeriveConcatSchema = z.object({
  op: z.literal('concat'),
  parts: z.array(z.string()).min(1).describe('字符串或 ${field} 引用'),
});

export const DeriveLowerSchema = z.object({
  op: z.literal('lower'),
  from: z.string().min(1),
});

export const DeriveUpperSchema = z.object({
  op: z.literal('upper'),
  from: z.string().min(1),
});

export const DeriveDefaultSchema = z.object({
  op: z.literal('default'),
  from: z.string().min(1),
  value: z.union([z.string(), z.number(), z.boolean()]),
});

export const DeriveTrimSchema = z.object({
  op: z.literal('trim'),
  from: z.string().min(1),
});

/// W4 末加:CJK→ASCII 标点 + trim 一站式归一(给中文宿主 enum_mapping 字符串匹配用)
/// trim 是子集(只去首尾噪音);normalize 还做"中文括号/逗号/分号 → ASCII"中段替换
/// 适用任何中文宿主的脏字段归一(不是 jvs-dw 特化),所以是通用 op
export const DeriveNormalizeSchema = z.object({
  op: z.literal('normalize'),
  from: z.string().min(1),
});

export const DeriveCoalesceSchema = z.object({
  op: z.literal('coalesce'),
  from: z.array(z.string().min(1)).min(1).describe('按顺序取第一个非空字段值'),
});

export const DeriveExprSchema = z.discriminatedUnion('op', [
  DeriveSubstringSchema,
  DeriveConcatSchema,
  DeriveLowerSchema,
  DeriveUpperSchema,
  DeriveDefaultSchema,
  DeriveTrimSchema,
  DeriveNormalizeSchema,
  DeriveCoalesceSchema,
]);
export type DeriveExpr = z.infer<typeof DeriveExprSchema>;

export const DeriveOpSchema = z.object({
  kind: z.literal('derive'),
  input: tableNameSchema,
  output: tableNameSchema,
  fields: z.record(z.string(), DeriveExprSchema),
});
export type DeriveOp = z.infer<typeof DeriveOpSchema>;

// ─────────────────────────────────────────────────────────
// ④ route_by_pattern — 按字段值分流到多 output 表
// ─────────────────────────────────────────────────────────

/**
 * 一份 input → N 份 output(按字段值规则分流)。
 *
 * routes:
 *   - 顺序匹配,第一个命中规则入对应 output 表
 *   - 都没命中:看是否有 `default: true` 的兜底分支;没有 → 丢弃 + warn 日志
 *
 * 规则:
 *   - starts_with: [...]   字段值前缀匹配任一
 *   - equals: [...]        字段值等于任一
 *   - contains: [...]      字段值包含任一子串
 *   - in: [...]            等同 equals(命名习惯)
 *   - default: true        兜底
 */
export const RouteRuleSchema = z.union([
  z.object({ starts_with: z.array(z.string()).min(1) }),
  z.object({ equals: z.array(z.string()).min(1) }),
  z.object({ contains: z.array(z.string()).min(1) }),
  z.object({ in: z.array(z.string()).min(1) }),
  z.object({ default: z.literal(true) }),
]);
export type RouteRule = z.infer<typeof RouteRuleSchema>;

export const RouteByPatternOpSchema = z.object({
  kind: z.literal('route_by_pattern'),
  input: tableNameSchema,
  field: z.string().min(1).describe('看哪个字段决定路由'),
  routes: z
    .array(
      z.object({
        output: tableNameSchema,
        when: RouteRuleSchema,
      }),
    )
    .min(1),
});
export type RouteByPatternOp = z.infer<typeof RouteByPatternOpSchema>;

// ─────────────────────────────────────────────────────────
// ⑤ pick_first_nonzero — 多列推断单列
// ─────────────────────────────────────────────────────────

/**
 * 例:fact_settlement_mode_out 有 17 个 pay_* 列(pay_wechat / pay_alipay / ...),
 * 按候选顺序找第一个 > 0 的,把对应 label 写到 output_field。
 *
 * 输入 row 不变,加新字段 output_field;没命中任何候选 → output_field = default。
 */
export const PickFirstNonzeroOpSchema = z.object({
  kind: z.literal('pick_first_nonzero'),
  input: tableNameSchema,
  output: tableNameSchema,
  output_field: z.string().min(1),
  default: z.string().default('other'),
  candidates: z
    .array(
      z.object({
        column: z.string().min(1).describe('数值列名'),
        value: z.string().min(1).describe('命中时填入 output_field 的 label'),
      }),
    )
    .min(1),
});
export type PickFirstNonzeroOp = z.infer<typeof PickFirstNonzeroOpSchema>;

// ─────────────────────────────────────────────────────────
// ⑥ filter — 行过滤(等价 SQL WHERE 行级过滤)
// ─────────────────────────────────────────────────────────

/**
 * 按 where 规则过滤行。where 是 AND 关系(所有规则必须通过),复用 WhereRuleSchema。
 *
 * 例(filter encounter 行,要求 in_time 非空):
 *   - kind: filter
 *     input: appointment_encounter_rows
 *     output: appointment_encounter_rows
 *     where:
 *       in_time: { not_empty: true }
 */
export const FilterOpSchema = z.object({
  kind: z.literal('filter'),
  input: tableNameSchema,
  output: tableNameSchema,
  where: z.record(z.string(), WhereRuleSchema),
});
export type FilterOp = z.infer<typeof FilterOpSchema>;

/**
 * union — 把多个同结构表合并成一个(行级 concat)。
 * 场景:treat_plan 与 plan 两源各自 route 出的"建议"行,合并进同一 recommendation 表。
 * inputs 缺失的表当空表;output 可与某 input 同名(in-place 合并)。
 */
export const UnionOpSchema = z.object({
  kind: z.literal('union'),
  inputs: z.array(tableNameSchema).min(1),
  output: tableNameSchema,
});
export type UnionOp = z.infer<typeof UnionOpSchema>;

// ─────────────────────────────────────────────────────────
// 7 operator union + transforms[] 顶层
// ─────────────────────────────────────────────────────────

export const TransformOpSchema = z.discriminatedUnion('kind', [
  ProjectOpSchema,
  SplitJsonArrayOpSchema,
  DeriveOpSchema,
  RouteByPatternOpSchema,
  PickFirstNonzeroOpSchema,
  FilterOpSchema,
  UnionOpSchema,
]);
export type TransformOp = z.infer<typeof TransformOpSchema>;

export const TransformsSchema = z.array(TransformOpSchema);
export type TransformsConfig = z.infer<typeof TransformsSchema>;
