import { BadRequestException, Injectable, Logger } from '@nestjs/common';
import { SyncDirection, SyncStatus } from '@pac/types';
import { PrismaService } from '../../../prisma/prisma.service';
import { PipelineDispatcher, type EmitsResolver } from '../pipeline/pipeline-dispatcher.service';
import { ColdImportService } from '../cold-import/cold-import.service';
import { QueueProducer } from '../../../queues/queue-producer.service';
import type { EmitsConfig } from '../assembler/assembler.schema';
import type { PushEvent, PushEventsResponse, PushRowsResponse } from './push.schema';

/**
 * PushReceiver — 处理已验签的 push 事件批。
 *
 * 流程:
 *   1. SyncLog start(direction=push, triggered_by='push:<host>:<first_source_event_id>')
 *   2. events → tables(按 subjectType 分组,把 payload + 元字段拍平为 canonical row)
 *   3. PipelineDispatcher 跑 dispatch
 *   4. SyncLog finalize + 返回 metrics
 *
 * **同步执行**:webhook 直接 ack metrics 给宿主。
 *   - 短期(目前)够用,体量 < 100 events/req
 *   - 后续接 BullMQ 异步化:本服务只做"验签 + 落 transaction + ack",parser 进队列
 */
@Injectable()
export class PushReceiverService {
  private readonly logger = new Logger(PushReceiverService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly dispatcher: PipelineDispatcher,
    private readonly coldImport: ColdImportService,
    private readonly queue: QueueProducer,
  ) {}

  /**
   * 形态 A:接收原生表行(/push/rows)。
   *
   * 复用 cold-import/reparse 同核(ColdImportService.ingestRawTables):transforms + assembler + 版本流落库,
   * 然后对 touched 患者入队【画像重算】(plan 不在此触发,由定时任务兜底)。
   */
  async receiveRows(input: {
    hostId: string;
    hostName: string;
    source: string;
    rows: Record<string, unknown>[];
  }): Promise<PushRowsResponse> {
    const { hostId, hostName, source, rows } = input;

    const r = await this.coldImport.ingestRawTables({
      hostName,
      source,
      rows,
      triggeredBy: `push:${hostName}:${source}`,
    });

    // 事件驱动:touched 患者入队 persona-recompute(BullMQ jobId 去重 + eventWatermark 幂等 + 凌晨 cron 兜底)
    // ⚠️ plan 不在此触发(交给现有定时任务),与单患者刷新路径有意区分。
    let personaEnqueued = 0;
    for (const { patientId, tenantId } of r.touched) {
      try {
        await this.queue.enqueuePersonaRecompute({
          hostId,
          tenantId,
          patientId,
          triggeredBy: `push:${hostName}`,
        });
        personaEnqueued++;
      } catch (err) {
        this.logger.warn(
          `enqueue persona-recompute failed patient=${patientId}: ${err instanceof Error ? err.message : err}`,
        );
      }
    }

    const agg = r.perResource.reduce(
      (a, s) => ({
        txn: a.txn + s.transactionsWritten,
        dup: a.dup + s.duplicates,
        failed: a.failed + s.failed,
      }),
      { txn: 0, dup: 0, failed: 0 },
    );

    this.logger.log(
      `push/rows host=${hostName} source=${source} rows=${rows.length} ` +
        `txn=${agg.txn} dup=${agg.dup} failed=${agg.failed} personaEnqueued=${personaEnqueued}`,
    );

    return {
      syncLogId: r.syncLogId,
      source,
      accepted: rows.length,
      transactionsWritten: agg.txn,
      duplicates: agg.dup,
      failed: agg.failed,
      personaEnqueued,
    };
  }

  async receive(input: {
    hostId: string;
    hostName: string;
    events: PushEvent[];
  }): Promise<PushEventsResponse> {
    const { hostId, hostName, events } = input;

    // 集团 = 事件携带的 tenantId(同登录授权口径);一批要求同属一个集团。
    const tenantId = events[0]?.tenantId ?? '';
    if (!tenantId || events.some((e) => e.tenantId !== tenantId)) {
      throw new BadRequestException(
        !tenantId ? 'push 事件缺 tenantId' : '同一批 push 事件必须同属一个集团(tenantId)',
      );
    }

    // host→tenant 一致性校验:tenantId 必须是该 host 已建立的集团之一。
    // 注:跨 host 隔离已由所有键的 host_id 保证(host 写不进别人数据);此处只防 host 自己
    // 填错 tenant slug 把数据散进新分区。新 host(尚无任何数据)首次放行,由本次写入建立其集团;
    // 接新集团请先经 cold-import / pull 建立,push 再跟上。
    const knownTenants = await this.prisma.patient.groupBy({
      by: ['tenantId'],
      where: { hostId },
    });
    if (knownTenants.length > 0 && !knownTenants.some((k) => k.tenantId === tenantId)) {
      throw new BadRequestException(
        `tenantId='${tenantId}' 不是 host=${hostName} 已建立的集团(已知 ` +
          `${knownTenants.map((k) => k.tenantId).join(' / ')});新集团请先经 cold-import / pull 建立`,
      );
    }

    const triggeredBy = `push:${hostName}:${events[0]?.subjectId ?? 'unknown'}`;

    // ─── 1. SyncLog start ───
    const syncLog = await this.prisma.syncLog.create({
      data: {
        hostId,
        tenantId,
        direction: SyncDirection.PUSH,
        resource: 'push_batch',
        triggeredBy,
        status: SyncStatus.RUNNING,
      },
    });

    // ─── 2. events → tables(按 subjectType 分组,canonicalRow 自带元字段 _action/_subjectType) ───
    const tables = this.eventsToTables(events);

    // ─── 3. dispatch(emits resolver:per-row,从行的 _action / _subjectType 元字段读) ───
    let metrics;
    let firstError: string | null = null;
    try {
      metrics = await this.dispatcher.dispatchTables({
        tables,
        hostId,
        tenantId,
        sourceContext: `push:${hostName}`,
        syncLogId: syncLog.id,
        emitsResolver: this.buildEmitsResolver(),
      });
    } catch (err) {
      firstError = err instanceof Error ? err.message : String(err);
      metrics = {
        patientsUpserted: 0,
        transactionsWritten: 0,
        duplicates: 0,
        failed: events.length,
        factsCreated: 0,
        factsSuperseded: 0,
        factsUnchanged: 0,
        factsFailed: 0,
      };
    }

    // ─── 4. finalize ───
    const status =
      firstError && metrics.transactionsWritten === 0
        ? SyncStatus.FAILED
        : firstError
          ? SyncStatus.PARTIAL
          : SyncStatus.SUCCESS;

    await this.prisma.syncLog.update({
      where: { id: syncLog.id },
      data: {
        status,
        fetched: events.length,
        transactionsWritten: metrics.transactionsWritten,
        factsEmitted: metrics.factsCreated + metrics.factsSuperseded,
        duplicates: metrics.duplicates,
        failed: metrics.failed,
        errorMessage: firstError,
        endedAt: new Date(),
      },
    });

    this.logger.log(
      `push received host=${hostName} events=${events.length} status=${status} ` +
        `txn=${metrics.transactionsWritten} dups=${metrics.duplicates} failed=${metrics.failed}`,
    );

    return {
      syncLogId: syncLog.id,
      accepted: events.length,
      transactionsWritten: metrics.transactionsWritten,
      duplicates: metrics.duplicates,
      failed: metrics.failed,
    };
  }

  /**
   * 把 events[] 转换成 PipelineDispatcher 需要的 tables 形态。
   *
   * 约定:event.subjectType 决定 table key(多元化:patient → patients,appointment → appointments)。
   * 每个 canonical row 由 event.payload + 元字段(externalId / clinicId / occurredAt 等)合成。
   */
  private eventsToTables(events: PushEvent[]): Record<string, unknown[]> {
    const tables: Record<string, unknown[]> = {};

    for (const e of events) {
      const tableKey = pluralizeSubject(e.subjectType);
      if (!tables[tableKey]) tables[tableKey] = [];

      // canonical row = payload + 元字段拍平
      const canonicalRow: Record<string, unknown> = {
        ...e.payload,
        externalId: e.subjectId,
        clinicId: e.clinicId,
        occurredAt: e.occurredAt,
        // 源组织命名空间来自事件信封 → dispatcher 优先取它(防跨命名空间撞号)
        sourceUnit: e.sourceUnit ?? '',
        ...(e.patientId ? { patientExternalId: e.patientId } : {}),
        // 末次修改时间 → synthesizer 合幂等键(内容变升版本)
        ...(e.updatedAt ? { updatedAt: e.updatedAt } : {}),
        // 给 synthesizer / dispatcher 用的元字段
        _action: e.action,
        _subjectType: e.subjectType,
        // ⭐ 真原文留底:host event payload(不含 PAC 加的 meta 字段)
        __source_row: e.payload,
      };
      tables[tableKey]!.push(canonicalRow);
    }

    return tables;
  }

  /**
   * emits resolver — push 路径 per-row 提取 emits(同 batch 不同 event 可能 action 不同)。
   *
   * 从 canonicalRow 元字段读:
   *   - _action       → emits.action
   *   - _subjectType  → emits.subjectType
   *   - occurredAtField 固定 = 'occurredAt'(push 协议 canonical 字段名)
   *
   * patient 主档(resource='patients')返回 undefined,走 dispatcher.upsertPatient 路径。
   */
  private buildEmitsResolver(): EmitsResolver {
    return (resource, row) => {
      if (resource === 'patients') return undefined;
      const action = (row._action as string | undefined) ?? '';
      const subjectType = (row._subjectType as string | undefined) ?? '';
      if (!action || !subjectType) {
        this.logger.warn(
          `push event 缺 _action / _subjectType 元字段;resource=${resource},跳过`,
        );
        return undefined;
      }
      return {
        action,
        subjectType,
        occurredAtField: 'occurredAt',
      } as EmitsConfig;
    };
  }
}

/**
 * subjectType('patient' / 'appointment') → tables key('patients' / 'appointments').
 * 对应 CanonicalResourceKey。
 */
function pluralizeSubject(subjectType: string): string {
  if (subjectType.endsWith('s')) return subjectType;  // 已是复数
  return subjectType + 's';
}
