import {
  Body,
  Controller,
  Headers,
  HttpCode,
  Post,
  Query,
  Req,
} from '@nestjs/common';
import type { Request } from 'express';
import { ApiOperation, ApiTags } from '@nestjs/swagger';
import { ApiCode } from '@pac/types';
import { BizError } from '../../../common/errors/biz-error';
import { Public } from '../../../common/decorators/public.decorator';
import { HmacVerifier } from './hmac-verifier.service';
import { PushReceiverService } from './push-receiver.service';
import {
  PushEventsRequestSchema,
  type PushEventsRequest,
  PushRowsRequestSchema,
  type PushRowsRequest,
} from './push.schema';

/**
 * PushController — 宿主主动推送事件入口(webhook)。
 *
 * 端点:POST /pac/v1/push/events
 *
 * Headers(必填):
 *   X-PAC-Host-Id: <uuid>
 *   X-PAC-Timestamp: <unix-seconds>
 *   X-PAC-Signature: <hex hmac-sha256(timestamp + "." + body, secret)>
 *
 * Body: { events: [{ subjectType, subjectRef, patientRef?, clinicId, action, sourceEventId, occurredAt, payload }] }
 *
 * 鉴权走 @Public + HMAC,不走 JWT(宿主推送不是用户登录)。
 *
 * **挂上 rawBody parser**:NestJS 默认 Express body-parser 会 JSON.parse 掉 body,
 * 但 HMAC 必须用原始 raw bytes 验签。需要在 main.ts 配 express.raw 或 verify 钩子保留 raw body。
 * 当前实现:从 req.rawBody 取(若 main.ts 已配);未配则 fallback 到 JSON.stringify(body) — 字段顺序不一致会 break。
 */
@ApiTags('push')
@Controller('push')
export class PushController {
  constructor(
    private readonly verifier: HmacVerifier,
    private readonly receiver: PushReceiverService,
  ) {}

  /**
   * 同 host 摄入并发闸(/events + /rows 共享)—— 磨平存量/增量的体制约束之一:
   * push 在服务进程内同步跑管线,存量重放是成百上千个批次连续打;并发闸 + 限批
   * = 天然限流,不与线上用户请求抢连接池。
   *
   * 额度 PAC_PUSH_CONCURRENCY(默认 4)。幂等 + updatedAt 乱序裁决保证并发与串行语义
   * 完全一致(宿主串行推也行,只是慢);额度守住连接池(4 批 × 批内串行写 ≈ 数个连接)。
   * 契约:超额并发返 CLIENT_RATE_LIMITED(10003),宿主退避重发即可。
   * 单进程内存计数即可(服务单实例部署;将来多实例再升 Redis 信号量)。
   */
  private readonly inflightByHost = new Map<string, number>();

  /** 手动 zod parse → 契约码 10002(裸 ZodError 会被兜成 90000,与文档"字段校验失败=10002"不符) */
  private parseOrReject<T>(parse: () => T): T {
    try {
      return parse();
    } catch (err) {
      throw new BizError(
        ApiCode.CLIENT_VALIDATION_FAILED,
        `push body 校验失败:${err instanceof Error ? err.message.slice(0, 300) : String(err)}`,
      );
    }
  }

  private async withHostLock<T>(hostId: string, hostName: string, fn: () => Promise<T>): Promise<T> {
    const limit = Math.max(1, Number(process.env.PAC_PUSH_CONCURRENCY) || 4);
    const cur = this.inflightByHost.get(hostId) ?? 0;
    if (cur >= limit) {
      throw new BizError(
        ApiCode.CLIENT_RATE_LIMITED,
        `host=${hostName} 并发批次已达上限(${limit});请退避后重发本批`,
      );
    }
    this.inflightByHost.set(hostId, cur + 1);
    try {
      return await fn();
    } finally {
      const n = (this.inflightByHost.get(hostId) ?? 1) - 1;
      if (n <= 0) this.inflightByHost.delete(hostId);
      else this.inflightByHost.set(hostId, n);
    }
  }

  @Post('events')
  @Public()
  @HttpCode(200)
  @ApiOperation({
    summary: '宿主推送事件 webhook(HMAC 验签)',
    description: 'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。',
  })
  async receive(
    @Req() req: Request & { rawBody?: Buffer },
    @Headers('x-pac-host-id') hostIdHeader: string | undefined,
    @Headers('x-pac-timestamp') timestampHeader: string | undefined,
    @Headers('x-pac-signature') signatureHeader: string | undefined,
    @Body() body: PushEventsRequest,
  ) {
    // raw body 优先(main.ts 应配 rawBody preserve);否则用 JSON.stringify(已 parsed)兜底
    const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body);

    const { hostId, hostName } = await this.verifier.verify({
      hostIdHeader,
      timestampHeader,
      signatureHeader,
      rawBody,
    });

    // body 单独走 zod 校验(手动 parse → 契约码 10002)
    // 集团 tenantId 由事件携带(receiver 内解析校验),不再由 host 名兜底。
    const parsed = this.parseOrReject(() => PushEventsRequestSchema.parse(body));

    return this.withHostLock(hostId, hostName, () =>
      this.receiver.receive({
        hostId,
        hostName,
        events: parsed.events,
      }),
    );
  }

  /**
   * 形态 A:推原生表行(推荐路径)。
   *
   * Body: { tenantId?, source, rows: [{原生字段照抄}, ...] }
   * 复用 pull/cold-import 同一条摄入管线(transforms + assembler + 版本流),
   * 落库后对 touched 患者入队画像重算(plan 由定时任务兜底)。
   */
  @Post('rows')
  @Public()
  @HttpCode(200)
  @ApiOperation({
    summary: '宿主推送原生表行 webhook(形态 A,HMAC 验签;?dryRun=1 预检不落库)',
    description:
      'Headers 必填 X-PAC-Host-Id / X-PAC-Timestamp / X-PAC-Signature。Body: { source, rows[] }。' +
      ' 加 ?dryRun=1 走**预检**:跑完整管线(transforms+装配+归一)但不落库、不触发重算,' +
      '响应带各资源样本 canonical —— 对接自测先预检到 failed=0 且 mappingMisses/suspectFields 为 0,再正式推。',
  })
  async receiveRows(
    @Req() req: Request & { rawBody?: Buffer },
    @Headers('x-pac-host-id') hostIdHeader: string | undefined,
    @Headers('x-pac-timestamp') timestampHeader: string | undefined,
    @Headers('x-pac-signature') signatureHeader: string | undefined,
    @Query('dryRun') dryRunQuery: string | undefined,
    @Body() body: PushRowsRequest,
  ) {
    const rawBody = req.rawBody ? req.rawBody.toString('utf8') : JSON.stringify(body);

    const { hostId, hostName } = await this.verifier.verify({
      hostIdHeader,
      timestampHeader,
      signatureHeader,
      rawBody,
    });

    const parsed = this.parseOrReject(() => PushRowsRequestSchema.parse(body));

    // ?dryRun=1 / true / yes 均视为预检(宿主客户端拼 query 习惯不一)
    const dryRun = ['1', 'true', 'yes'].includes((dryRunQuery ?? '').trim().toLowerCase());

    return this.withHostLock(hostId, hostName, () =>
      this.receiver.receiveRows({
        hostId,
        hostName,
        source: parsed.source,
        rows: parsed.rows,
        dryRun,
      }),
    );
  }
}
