import { 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 type { EmitsConfig } from '../assembler/assembler.schema';
import type { PushEvent, PushEventsResponse } 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,
  ) {}

  async receive(input: {
    hostId: string;
    hostName: string;
    tenantId: string;
    events: PushEvent[];
  }): Promise<PushEventsResponse> {
    const { hostId, hostName, tenantId, events } = input;
    const firstSourceId = events[0]?.sourceEventId ?? 'unknown';
    const triggeredBy = `push:${hostName}:${firstSourceId}`;

    // ─── 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.subjectRef,
        clinicId: e.clinicId,
        occurredAt: e.occurredAt,
        ...(e.patientRef ? { patientExternalId: e.patientRef } : {}),
        // 给 synthesizer 用的元字段
        _sourceEventId: e.sourceEventId,
        _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';
}
