import { Injectable, Logger } from '@nestjs/common';
import { runProject } from './operators/project.op';
import { runSplitJsonArray } from './operators/split-json-array.op';
import { runDerive } from './operators/derive.op';
import { runRouteByPattern } from './operators/route-by-pattern.op';
import { runPickFirstNonzero } from './operators/pick-first-nonzero.op';
import { runFilter } from './operators/filter.op';
import { SOURCE_ROW_KEY, type TableMap } from './row';
import type { TransformsConfig } from './transforms.schema';

/**
 * TransformEngine — yaml 声明式 transform pipeline 执行器(Layer A.5)。
 *
 * 输入:source tables(Record<tableName, Row[]>) + transforms[](顺序操作)
 * 输出:扩展后的 tables map(包含 source 表 + transform 产出的新表)
 *
 * 纪律(canonical-fact-layer.md 原则 1):5 operator 白名单,撞墙必须加 op + PR。
 *
 * 一致性保证:
 *   - 输入表不可变(每个 op 拷贝输出,不污染上游)
 *   - 中间表名通常以 `_` 开头(惯例),最终消费表名跟 assembler.primary.table 对齐
 *   - 同名 output 表:后写覆盖,允许 in-place derive(`output: input`)
 */
@Injectable()
export class TransformEngine {
  private readonly logger = new Logger(TransformEngine.name);

  /**
   * 跑 transforms[] 序列,返回扩展后的 tables map。
   *
   * **入口**:给所有源 table 的行打 `__source_row = 原 host row` 标记,
   *   后续 operator 透传,AssemblerEngine 出口提取作真原文留底(raw_payload)。
   */
  run(opts: { tables: TableMap; transforms: TransformsConfig }): TableMap {
    const tables: TableMap = {};
    // 入口标记:每个源行带上自身的 __source_row 引用(深拷贝避免后续 operator 修改污染)
    for (const [name, rows] of Object.entries(opts.tables)) {
      tables[name] = rows.map((row) => {
        // 已有 __source_row 的不覆盖(允许 caller 提前注入,比如 Push 路径)
        if (SOURCE_ROW_KEY in row) return row;
        return { ...row, [SOURCE_ROW_KEY]: row };
      });
    }

    for (const op of opts.transforms) {
      // union 是多输入(op.inputs),不走下面的单 input 取值;缺失表当空表
      if (op.kind === 'union') {
        const merged = op.inputs.flatMap((n) => tables[n] ?? []);
        tables[op.output] = merged;
        this.logger.log(
          `[transform/union] [${op.inputs.join(',')}] → "${op.output}"(${merged.length})`,
        );
        continue;
      }
      const inputName = op.input;
      const input = tables[inputName];
      if (!input) {
        this.logger.warn(
          `[transform/${op.kind}] input table "${inputName}" 不存在;跳过`,
        );
        continue;
      }

      switch (op.kind) {
        case 'project': {
          const out = runProject(op, input);
          tables[op.output] = out;
          this.logger.log(
            `[transform/project] "${inputName}"(${input.length}) → "${op.output}"(${out.length})`,
          );
          break;
        }
        case 'split_json_array': {
          const { rows, skippedParents, skippedChildren } = runSplitJsonArray(op, input);
          tables[op.output] = rows;
          this.logger.log(
            `[transform/split_json_array] "${inputName}"(${input.length}) → "${op.output}"(${rows.length})` +
              (skippedParents > 0 ? ` skipped_parents=${skippedParents}` : '') +
              (skippedChildren > 0 ? ` skipped_children=${skippedChildren}` : ''),
          );
          break;
        }
        case 'derive': {
          const out = runDerive(op, input);
          tables[op.output] = out;
          this.logger.log(
            `[transform/derive] "${inputName}"(${input.length}) → "${op.output}"(${out.length}) fields=${Object.keys(op.fields).join(',')}`,
          );
          break;
        }
        case 'route_by_pattern': {
          const { outputs, dropped } = runRouteByPattern(op, input);
          for (const [name, rows] of Object.entries(outputs)) {
            tables[name] = rows;
          }
          const counts = Object.entries(outputs)
            .map(([n, r]) => `${n}=${r.length}`)
            .join(',');
          this.logger.log(
            `[transform/route] "${inputName}"(${input.length}) → ${counts}` +
              (dropped > 0 ? ` dropped=${dropped}` : ''),
          );
          break;
        }
        case 'pick_first_nonzero': {
          const out = runPickFirstNonzero(op, input);
          tables[op.output] = out;
          this.logger.log(
            `[transform/pick_first_nonzero] "${inputName}"(${input.length}) → "${op.output}".${op.output_field}`,
          );
          break;
        }
        case 'filter': {
          const { rows, dropped } = runFilter(op, input);
          tables[op.output] = rows;
          this.logger.log(
            `[transform/filter] "${inputName}"(${input.length}) → "${op.output}"(${rows.length})` +
              (dropped > 0 ? ` dropped=${dropped}` : ''),
          );
          break;
        }
      }
    }

    return tables;
  }
}
