import { protocolUint32ToDb } from '../common/protocol-uint32';
import { completionContext } from './completion-context';
import { resolveUplinkMatch } from './uplink-matching';
import { BadRequestException, Logger, NotFoundException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { createHash, randomUUID } from 'node:crypto';
import { BillingService } from '../billing/billing.service';

import type { OpenApiService } from '../open-api/open-api.service';
import { PrismaService } from '../prisma/prisma.service';
import type {
  GatewayUplinkEventDto,
  UplinkMatchCandidateInput,
  GatewayControlDeliveryResult,
} from './send-chain.contracts';
import { downstreamControlFailureMessage } from './send-chain.helpers';

import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service';
import { queueFinalReceiptDeliveries, type DownstreamDeliveryQueueRequest } from './downstream-receipt-targets';

/**
 * R10 downstreamDelivery implementation.
 * Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability.
 */
export class SendDownstreamDeliveryService {
  private readonly logger = new Logger('SendChainService');

  constructor(
    private readonly prisma: PrismaService,
    private readonly billing: BillingService,
    private readonly openApi: OpenApiService | undefined,
    private readonly facade: SendCompletionFacade,
    private readonly callbacks: SendCompletionCallbacks,
  ) {}

  async handleUplink(data: GatewayUplinkEventDto) {
    protocolUint32ToDb(data.sequenceId);
    if (!completionContext.getStore()) {
      return this.prisma.$transaction(
        (tx) => completionContext.run({ tx, messageRecordId: '' }, () => this.persistUplink(data)),
        { timeout: 15_000 },
      );
    }
    return this.persistUplink(data);
  }

  private async persistUplink(data: GatewayUplinkEventDto) {
    if (data.eventId) {
      await this.prisma
        .$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`uplink-event:${data.eventId}`},0))`;
      const existing = await this.prisma.smsUplinkMessage.findUnique({ where: { eventId: data.eventId } });
      if (existing) return existing;
    }
    const channel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
    if (!channel) {
      throw new NotFoundException('SMS channel not found');
    }
    const match = await this.facade.resolveUplinkMatch(data, channel);
    const record = await this.prisma.smsUplinkMessage.create({
      data: {
        eventId: data.eventId,
        tenantId: match.tenantId,
        applicationId: match.applicationId,
        messageRecordId: match.messageRecordId,
        channelId: data.channelId,
        messageId: match.messageId,
        gatewayMessageId: data.gatewayMessageId,
        sequenceId: protocolUint32ToDb(data.sequenceId),
        phoneNumber: data.phoneNumber,
        destId: data.destId,
        content: data.content,
        matchStatus: match.matchStatus,
        matchReason: match.matchReason,
        receivedAt: data.receivedAt ? new Date(data.receivedAt) : new Date(),
      },
    });
    if (match.candidates.length > 0) {
      await this.prisma.smsUplinkMatchCandidate.createMany({
        data: match.candidates.map((candidate) => ({
          uplinkMessageId: record.id,
          tenantId: candidate.tenantId,
          applicationId: candidate.applicationId,
          messageRecordId: candidate.messageRecordId,
          matchSource: candidate.matchSource,
          confidence: candidate.confidence,
          reason: candidate.reason,
        })),
        skipDuplicates: true,
      });
    }
    if (match.tenantId && match.applicationId) {
      await this.facade.queueAndTryDownstreamDelivery({
        tenantId: match.tenantId,
        applicationId: match.applicationId,
        messageRecordId: match.messageRecordId,
        messageId: match.messageId,
        deliveryType: 'uplink',
        payload: {
          messageId: match.messageId,
          applicationId: match.applicationId,
          phoneNumber: data.phoneNumber,
          destId: data.destId,
          content: data.content,
          receivedAt: record.receivedAt.toISOString(),
          uplinkMessageId: record.id,
        },
      });
    }
    return record;
  }

  async claimUplinkMatchCandidate(uplinkMessageId: string, candidateId: string, operatorId?: string) {
    if (!completionContext.getStore()) {
      return this.prisma.$transaction(
        (tx) =>
          completionContext.run({ tx, messageRecordId: '' }, () =>
            this.persistUplinkClaim(uplinkMessageId, candidateId, operatorId),
          ),
        { timeout: 15_000 },
      );
    }
    return this.persistUplinkClaim(uplinkMessageId, candidateId, operatorId);
  }

  private async persistUplinkClaim(uplinkMessageId: string, candidateId: string, operatorId?: string) {
    await this.prisma
      .$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`uplink-claim:${uplinkMessageId}`},0))`;
    const candidate = await this.prisma.smsUplinkMatchCandidate.findFirst({
      where: { id: candidateId, uplinkMessageId },
      include: {
        application: { select: { id: true, name: true, cmppAccount: true } },
        messageRecord: { select: { id: true, messageId: true, content: true } },
        uplinkMessage: true,
      },
    });
    if (!candidate) {
      throw new NotFoundException('Uplink match candidate not found');
    }
    if (candidate.status === 'rejected') {
      throw new BadRequestException('该候选已被排除，不能认领');
    }
    if (candidate.uplinkMessage.matchStatus === 'matched' && candidate.status !== 'claimed') {
      throw new BadRequestException('该上行记录已完成匹配，不能重复认领');
    }
    if (candidate.status === 'claimed') return candidate.uplinkMessage;
    const claimedAt = new Date();
    const messageId = candidate.messageRecord?.messageId ?? null;
    const updatedUplink = await this.prisma.smsUplinkMessage.update({
      where: { id: uplinkMessageId },
      data: {
        tenantId: candidate.tenantId,
        applicationId: candidate.applicationId,
        messageRecordId: candidate.messageRecordId,
        messageId,
        matchStatus: 'matched',
        matchReason: `人工认领：${candidate.reason ?? candidate.matchSource}`,
      },
    });
    await this.prisma.smsUplinkMatchCandidate.updateMany({
      where: {
        uplinkMessageId,
        id: { not: candidate.id },
        status: 'pending',
      },
      data: { status: 'rejected' },
    });
    await this.prisma.smsUplinkMatchCandidate.update({
      where: { id: candidate.id },
      data: {
        status: 'claimed',
        claimedAt,
        claimedById: operatorId,
      },
    });
    await this.prisma.operationLog.create({
      data: {
        tenantId: candidate.tenantId,
        userId: operatorId,
        action: 'gateway.uplink_manual_claim',
        resource: 'sms_uplink_message',
        resourceId: uplinkMessageId,
        detail: {
          candidateId: candidate.id,
          applicationId: candidate.applicationId,
          applicationName: candidate.application.name,
          messageRecordId: candidate.messageRecordId,
          messageId,
          matchSource: candidate.matchSource,
          phoneNumber: candidate.uplinkMessage.phoneNumber,
          destId: candidate.uplinkMessage.destId,
        },
      },
    });

    await this.facade.queueAndTryDownstreamDelivery({
      tenantId: candidate.tenantId,
      applicationId: candidate.applicationId,
      messageRecordId: candidate.messageRecordId,
      messageId,
      deliveryType: 'uplink',
      payload: {
        messageId,
        applicationId: candidate.applicationId,
        phoneNumber: candidate.uplinkMessage.phoneNumber,
        destId: candidate.uplinkMessage.destId,
        content: candidate.uplinkMessage.content,
        receivedAt: candidate.uplinkMessage.receivedAt.toISOString(),
        manualClaim: true,
        uplinkMessageId,
      },
    });
    return this.prisma.smsUplinkMessage.findUnique({
      where: { id: updatedUplink.id },
      include: {
        tenant: true,
        application: true,
        channel: true,
        messageRecord: { include: { application: true } },
        matchCandidates: {
          include: {
            tenant: true,
            application: true,
            messageRecord: { include: { application: true, tenant: true, channel: true } },
          },
          orderBy: [{ status: 'asc' }, { confidence: 'desc' }, { createdAt: 'asc' }],
        },
      },
    });
  }

  async queueAndTryDownstreamDelivery(data: DownstreamDeliveryQueueRequest) {
    if (!data.applicationId) {
      return null;
    }
    const application = await this.prisma.smsApplication.findUnique({
      where: { id: data.applicationId },
      select: {
        cmppAccount: true,
        interfaceEnabled: true,
        status: true,
        downstreamReceiptRetryEnabled: true,
        downstreamUplinkRetryEnabled: true,
        httpConfig: true,
      },
    });
    const deliveryAllowed = application?.status === 'active' || application?.status === 'disabling';
    const cmppDeliveryAllowed = deliveryAllowed || data.allowBusinessRejectionCmppDelivery === true;
    if (deliveryAllowed && data.queueHttpWebhook !== false) {
      try {
        await this.openApi?.queueWebhookEvent(
          {
            tenantId: data.tenantId,
            applicationId: data.applicationId,
            messageRecordId: data.messageRecordId,
            messageId: data.messageId,
            uplinkMessageId:
              typeof data.payload.uplinkMessageId === 'string' ? data.payload.uplinkMessageId : undefined,
            eventType: data.deliveryType,
            payload: data.payload,
          },
          completionContext.getStore()?.tx,
        );
      } catch (error) {
        this.logger.error(
          `HTTP webhook queue failed for ${data.deliveryType}/${data.messageId ?? '-'}: ${error instanceof Error ? error.message : String(error)}`,
        );
        if (data.propagateHttpQueueError || completionContext.getStore()) throw error;
      }
    }
    if (data.queueCmppDelivery === false) {
      return null;
    }
    if (
      !application?.cmppAccount ||
      (application.interfaceEnabled !== true && data.allowBusinessRejectionCmppDelivery !== true)
    ) {
      return null;
    }
    const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload };
    const dedupeKey =
      data.deliveryType === 'receipt' && data.messageRecordId
        ? (data.receiptDedupeKey ?? `receipt:${data.messageRecordId}`)
        : data.deliveryType === 'uplink' && typeof data.payload.uplinkMessageId === 'string'
          ? `uplink:${data.payload.uplinkMessageId}`
          : null;
    if (completionContext.getStore()) {
      if (!dedupeKey) throw new Error('completion_notification_identity_missing');
      await this.prisma.cmppDownstreamDelivery.createMany({
        data: [
          {
            tenantId: data.tenantId,
            applicationId: data.applicationId,
            messageRecordId: data.messageRecordId,
            messageId: data.messageId,
            dedupeKey,
            deliveryType: data.deliveryType,
            payload,
            retryEnabled:
              cmppDeliveryAllowed &&
              (data.deliveryType === 'uplink'
                ? application.downstreamUplinkRetryEnabled
                : application.downstreamReceiptRetryEnabled),
            status: cmppDeliveryAllowed ? 'pending' : 'abandoned',
            lastError: cmppDeliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
          },
        ],
        skipDuplicates: true,
      });
      const retained = await this.prisma.cmppDownstreamDelivery.findUniqueOrThrow({ where: { dedupeKey } });
      if (
        (retained.messageRecordId ?? null) !== (data.messageRecordId ?? null) ||
        retained.applicationId !== data.applicationId
      )
        throw new Error('completion_notification_identity_mismatch');
      return retained;
    }
    let delivery;
    try {
      delivery = await this.prisma.cmppDownstreamDelivery.create({
        data: {
          tenantId: data.tenantId,
          applicationId: data.applicationId,
          messageRecordId: data.messageRecordId,
          messageId: data.messageId,
          dedupeKey,
          deliveryType: data.deliveryType,
          payload,
          retryEnabled:
            cmppDeliveryAllowed &&
            (data.deliveryType === 'uplink'
              ? (application?.downstreamUplinkRetryEnabled ?? true)
              : (application?.downstreamReceiptRetryEnabled ?? true)),
          status: cmppDeliveryAllowed ? 'pending' : 'abandoned',
          lastError: cmppDeliveryAllowed ? null : '企业应用已停用，保留回执但不再向客户应用推送',
        },
      });
    } catch (error) {
      if (dedupeKey && error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002') {
        const existing = await this.prisma.cmppDownstreamDelivery.findUnique({
          where: { dedupeKey },
        });
        if (existing) {
          this.logger.warn(
            `downstream_delivery_deduplicated ${JSON.stringify({
              deliveryType: data.deliveryType,
              messageRecordId: data.messageRecordId,
              messageId: data.messageId,
              dedupeKey,
              deliveryId: existing.id,
            })}`,
          );
          return existing;
        }
      }
      throw error;
    }
    if (!cmppDeliveryAllowed) {
      return delivery;
    }
    const claimId = `api-direct:${process.pid}:${randomUUID()}`;
    const claim = await this.prisma.cmppDownstreamDelivery.updateMany({
      where: { id: delivery.id, status: 'pending' },
      data: {
        status: 'dispatching',
        connectionId: claimId,
        ackDeadlineAt: new Date(Date.now() + 30_000),
        nextRetryAt: null,
        lastError: null,
      },
    });
    if (claim.count !== 1) {
      return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } });
    }
    try {
      const result = (await this.facade.postGatewayControl(
        data.deliveryType === 'receipt' ? '/downstream/receipt' : '/downstream/uplink',
        { deliveryId: delivery.id, claimId, ...payload },
      )) as GatewayControlDeliveryResult;
      if (result.sent || result.delivered) {
        await this.facade.markDownstreamDeliverySent({ id: delivery.id, claimId, ...result });
      } else if (result.reasonCode === 'SUBMIT_RESPONSE_PENDING') {
        await this.facade.markDownstreamDeliveryFailed(
          delivery.id,
          downstreamControlFailureMessage(result),
          'claim_released',
          { id: delivery.id, claimId, ...result },
        );
      } else {
        await this.facade.markDownstreamDeliveryFailed(
          delivery.id,
          downstreamControlFailureMessage(result),
          result.retryable === false ? 'unrecoverable' : 'send_failed',
          { id: delivery.id, claimId, ...result },
        );
      }
    } catch (error) {
      await this.facade.markDownstreamDeliveryFailed(
        delivery.id,
        error instanceof Error ? error.message : 'Gateway control delivery failed',
      );
    }
    return delivery;
  }

  async resolveUplinkMatch(
    data: GatewayUplinkEventDto,
    channel: { id: string; srcId?: string | null },
  ): Promise<{
    tenantId?: string;
    applicationId?: string;
    messageRecordId?: string;
    messageId?: string;
    matchStatus: string;
    matchReason: string;
    candidates: UplinkMatchCandidateInput[];
  }> {
    return resolveUplinkMatch(this.prisma, data, channel);
  }

  async recordCmppFailureReceipt(
    message: {
      id: string;
      tenantId?: string | null;
      batchTaskId?: string | null;
      applicationId?: string | null;
      messageId: string;
      phoneNumber: string;
      cmppSubmitSequenceId?: string | null;
      cmppSubmitGroupMessageId?: string | null;
      cmppRegisteredDelivery?: boolean | null;
    },
    errorCode: string,
    reason: string,
  ) {
    if (!message.tenantId || !message.applicationId) return null;
    const existing = await this.prisma.smsReceiptRecord.findFirst({
      where: { messageRecordId: message.id, gatewayMessageId: `PLATFORM:${message.messageId}` },
    });
    const recoverableRejection = errorCode.startsWith('DRN') || errorCode === 'CSW';
    if (existing && !recoverableRejection) return existing;
    const deliveredAt = new Date();
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: {
        status: 'failed',
        receiptStatus: 'undelivered',
        receiptRawStatus: 'REJECTD',
        errorCode,
        errorMessage: reason,
        deliveredAt,
      },
    });
    const gatewayMessageId = `PLATFORM:${message.messageId}`;
    const receiptKey = createHash('sha256')
      .update(
        `platform\u0000${gatewayMessageId}\u0000${message.phoneNumber}\u0000undelivered\u0000REJECTD\u0000${errorCode}`,
      )
      .digest('hex');
    const receiptData = {
      tenantId: message.tenantId,
      batchTaskId: message.batchTaskId,
      messageRecordId: message.id,
      receiptKey,
      messageId: message.messageId,
      gatewayMessageId,
      phoneNumber: message.phoneNumber,
      receiptStatus: 'undelivered',
      rawStatus: 'REJECTD',
      errorCode,
      errorMessage: reason,
      deliveredAt,
    };
    const receipt =
      existing ??
      (recoverableRejection
        ? await this.prisma.smsReceiptRecord.upsert({ where: { receiptKey }, update: {}, create: receiptData })
        : await this.prisma.smsReceiptRecord.create({ data: receiptData }));
    await queueFinalReceiptDeliveries(this.prisma, (request) => this.facade.queueAndTryDownstreamDelivery(request), {
      message,
      propagateHttpQueueError: recoverableRejection,
      allowBusinessRejectionCmppDelivery: errorCode === 'ACCOUNT' || errorCode === 'INTERFACE',
      payload: {
        messageId: message.messageId,
        gatewayMessageId: `PLATFORM:${message.messageId}`,
        phoneNumber: message.phoneNumber,
        receiptStatus: 'undelivered',
        rawStatus: 'REJECTD',
        errorCode,
        errorMessage: reason,
        deliveredAt: deliveredAt.toISOString(),
      },
    });
    if (message.batchTaskId) await this.facade.refreshTaskProgress(message.batchTaskId);
    if (errorCode === 'CSW')
      await this.prisma.smsMessageRecord.update({
        where: { id: message.id },
        data: { channelWordFinalizationPending: false },
      });
    if (errorCode.startsWith('DRN'))
      await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { drainageReceiptPending: false } });
    return receipt;
  }

  async postGatewayControl(path: string, payload: unknown) {
    const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? 'http://127.0.0.1:8090').replace(/\/$/, '');
    const response = await fetch(`${baseUrl}${path}`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(payload),
    });
    if (!response.ok) {
      const body = await response.text().catch(() => '');
      throw new Error(`Gateway control ${path} returned ${response.status}${body ? `: ${body}` : ''}`);
    }
    return response.json().catch(() => ({}));
  }
}