import { Injectable, Logger, NotFoundException } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { SyncDirection, SyncStatus, type PullConfig } from '@pac/types';
import { PrismaService } from '../../../prisma/prisma.service';
import { PullStrategyRegistry } from '../adapters/adapter.registry';
import { PipelineDispatcher, type DispatchMetrics } from '../pipeline/pipeline-dispatcher.service';
import type { SyncContext } from '../adapters/adapter.interface';

/**
 * ReconcileOrchestrator — 对账(spec § 七层防线 #4)。
 *
 * 跟 PullOrchestrator 关键区别:
 *   1. 按**时间窗口**拉(过去 N 天),不走 cursor
 *   2. **不推进** SyncLog.cursorAfter(避免污染正常 pull 水位)
 *   3. triggered_by='reconcile:<date>'(便于审计区分)
 *   4. 写入靠 partial UNIQUE(source_event_id) 自动去重 — 缺什么补什么
 *
 * 用途:补 push 丢失、pull cursor 漏拉、PAC 故障窗口期数据。
 * 即使 PAC 挂 7 天,恢复后跑一次 reconcile 也能追平。
 *
 * 优先 strategy.pullWindow();不支持则退化为 cursor=null 反复 pullPage(吃所有数据 + 去重)
 */
@Injectable()
export class ReconcileOrchestrator {
  private readonly logger = new Logger(ReconcileOrchestrator.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly registry: PullStrategyRegistry,
    private readonly dispatcher: PipelineDispatcher,
  ) {}

  async runForHost(
    hostId: string,
    options: { windowDays?: number; triggeredBy?: string } = {},
  ): Promise<ReconcileRunResult> {
    const host = await this.prisma.host.findUnique({ where: { id: hostId } });
    if (!host) throw new NotFoundException(`Host ${hostId} not found`);
    if (!host.active || !host.pullConfig) {
      return { skipped: true, reason: !host.active ? 'host_inactive' : 'no_pull_config' };
    }

    const tenantId = host.name.includes('demo') || host.name.includes('mock') ? 't_demo' : 't_default';
    const windowDays = options.windowDays ?? 7;
    const endedAt = new Date();
    const startedAt = new Date(endedAt.getTime() - windowDays * 86400_000);
    const dateTag = endedAt.toISOString().slice(0, 10);
    const runId = randomUUID();
    const triggeredBy = options.triggeredBy ?? `reconcile:${dateTag}`;

    // ─── SyncLog start ───
    const syncLog = await this.prisma.syncLog.create({
      data: {
        hostId: host.id,
        tenantId,
        direction: SyncDirection.PULL,
        resource: 'reconcile_run',
        triggeredBy,
        status: SyncStatus.RUNNING,
        // 故意不读不写 cursorBefore/cursorAfter
      },
    });

    // ─── 调 strategy ───
    const ctx: SyncContext = {
      hostId: host.id,
      tenantId,
      pullConfig: host.pullConfig as PullConfig,
      runId,
    };
    const strategy = this.registry.resolve(ctx);
    if (strategy.prepare) await strategy.prepare(ctx);

    const totals = zeroDispatchMetrics();
    let pages = 0;
    let firstError: string | null = null;

    try {
      // 优先 pullWindow(适配器知道怎么按时间窗高效查);否则降级到 pullPage 全量
      if (strategy.pullWindow) {
        for await (const page of strategy.pullWindow(ctx, startedAt, endedAt)) {
          pages++;
          const isEmpty = Object.values(page.tables).every((t) => t.length === 0);
          if (!isEmpty) {
            const emitsPerResource = page.emitsPerResource ?? {};
            const metrics = await this.dispatcher.dispatchTables({
              tables: page.tables,
              hostId: host.id,
              tenantId,
              sourceContext: `reconcile:${host.name}`,
              syncLogId: syncLog.id,
              emitsResolver: (resource) => emitsPerResource[resource],
            });
            addMetrics(totals, metrics);
          }
        }
      } else {
        // 降级:无窗口能力,从头 paginate 全部(去重靠 partial UNIQUE)
        this.logger.warn(`host=${host.name} adapter 无 pullWindow,fallback paginate-all`);
        let cursor: string | null = null;
        while (true) {
          const page = await strategy.pullPage(ctx, cursor);
          pages++;
          const isEmpty = Object.values(page.tables).every((t) => t.length === 0);
          if (!isEmpty) {
            const emitsPerResource = page.emitsPerResource ?? {};
            const metrics = await this.dispatcher.dispatchTables({
              tables: page.tables,
              hostId: host.id,
              tenantId,
              sourceContext: `reconcile:${host.name}`,
              syncLogId: syncLog.id,
              emitsResolver: (resource) => emitsPerResource[resource],
            });
            addMetrics(totals, metrics);
          }
          if (!page.nextCursor) break;
          cursor = page.nextCursor;
        }
      }
    } catch (err) {
      firstError = err instanceof Error ? err.message : String(err);
      this.logger.error(`reconcile host=${host.name} 失败: ${firstError}`);
    }

    const status =
      firstError && totals.transactionsWritten === 0 ? SyncStatus.FAILED : firstError ? SyncStatus.PARTIAL : SyncStatus.SUCCESS;

    await this.prisma.syncLog.update({
      where: { id: syncLog.id },
      data: {
        status,
        fetched: totals.patientsUpserted + totals.transactionsWritten + totals.duplicates,
        transactionsWritten: totals.transactionsWritten,
        factsEmitted: totals.factsCreated + totals.factsSuperseded,
        duplicates: totals.duplicates,  // 关键:reconcile 大部分应该是 duplicates,缺失的会出现在 transactionsWritten
        failed: totals.failed,
        errorMessage: firstError,
        endedAt: new Date(),
      },
    });

    this.logger.log(
      `reconcile done host=${host.name} status=${status} window=${windowDays}d ` +
        `pages=${pages} new=${totals.transactionsWritten} dups=${totals.duplicates}(预期占大头) failed=${totals.failed}`,
    );

    return {
      skipped: false,
      syncLogId: syncLog.id,
      status,
      windowDays,
      pages,
      metrics: totals,
    };
  }
}

export type ReconcileRunResult =
  | { skipped: true; reason: string }
  | {
      skipped: false;
      syncLogId: string;
      status: string;
      windowDays: number;
      pages: number;
      metrics: DispatchMetrics;
    };

function zeroDispatchMetrics(): DispatchMetrics {
  return {
    patientsUpserted: 0,
    transactionsWritten: 0,
    duplicates: 0,
    failed: 0,
    factsCreated: 0,
    factsSuperseded: 0,
    factsUnchanged: 0,
    factsFailed: 0,
    personaEnqueued: 0,
  };
}

function addMetrics(target: DispatchMetrics, add: DispatchMetrics): void {
  target.patientsUpserted += add.patientsUpserted;
  target.transactionsWritten += add.transactionsWritten;
  target.duplicates += add.duplicates;
  target.failed += add.failed;
  target.factsCreated += add.factsCreated;
  target.factsSuperseded += add.factsSuperseded;
  target.factsUnchanged += add.factsUnchanged;
  target.factsFailed += add.factsFailed;
  target.personaEnqueued += add.personaEnqueued;
}
