import { BadRequestException, Injectable, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
import { randomUUID } from 'node:crypto';
import { createHash } from 'node:crypto';
import { isIP } from 'node:net';
import { setTimeout as sleep } from 'node:timers/promises';
import { BillingService } from '../billing/billing.service';
import { PrismaService } from '../prisma/prisma.service';
import { RiskReviewService } from '../risk-review/risk-review.service';

export interface CreateBatchTaskDto {
  tenantId: string;
  applicationId?: string;
  templateId?: string;
  content: string;
  category?: string;
  phones: string[];
  sendMode?: 'immediate' | 'scheduled';
  scheduledAt?: string;
  variables?: Record<string, unknown>;
  createdById?: string;
  sourceIp?: string;
  userAgent?: string;
  sourceType?: 'client' | 'api' | 'cmpp';
}

export interface GatewayInboundAuthDto {
  account: string;
  password?: string;
  authSource?: string;
  timestamp?: number;
  remoteIp?: string;
}

export interface GatewayInboundSubmitDto {
  account: string;
  phoneNumber: string;
  content: string;
  srcId?: string;
  destId?: string;
  sequenceId?: number;
  remoteIp?: string;
}

export interface GatewaySubmitResultDto {
  traceId?: string;
  messageId: string;
  channelId: string;
  submitId?: string;
  sequenceId?: number;
  gatewayMessageId: string;
  submitStatus: 'accepted' | 'rejected' | 'timeout';
  errorCode?: string;
  errorMessage?: string;
  submittedAt?: string;
  segments?: Array<{
    segmentTotal?: number;
    segmentIndex?: number;
    sequenceId?: number;
    gatewayMessageId?: string;
    submitStatus?: 'accepted' | 'rejected' | 'timeout' | string;
    errorCode?: string;
    errorMessage?: string;
    submittedAt?: string;
  }>;
}

export interface GatewayReceiptEventDto {
  traceId?: string;
  messageId?: string;
  channelId: string;
  sequenceId?: number;
  gatewayMessageId: string;
  phoneNumber?: string;
  receiptStatus: 'delivered' | 'undelivered' | 'unknown';
  rawStatus: string;
  errorCode?: string;
  deliveredAt?: string;
}

export interface GatewayUplinkEventDto {
  traceId?: string;
  messageId?: string;
  channelId: string;
  sequenceId?: number;
  phoneNumber: string;
  destId: string;
  content: string;
  receivedAt?: string;
}

type UplinkMatchCandidateInput = {
  tenantId: string;
  applicationId: string;
  messageRecordId?: string;
  matchSource: 'access_number' | 'phone_window';
  confidence: number;
  reason: string;
};

export interface GatewayPendingDeliveryQueryDto {
  account: string;
  limit?: number;
}

export interface GatewayDownstreamSentDto {
  id: string;
  connectionId?: string;
  sequenceId?: string;
  messageId?: string;
  sentAt?: string;
  ackDeadlineAt?: string;
}

export interface GatewayDownstreamAcknowledgedDto extends GatewayDownstreamSentDto {
  result: number;
  acknowledgedAt?: string;
}

export type GatewayDownstreamFailureType = 'send_failed' | 'ack_timeout' | 'ack_rejected' | 'ack_invalid' | 'connection_lost';

type GatewayControlDeliveryResult = {
  sent?: boolean;
  delivered?: boolean;
  connectionId?: string;
  sequenceId?: string;
  messageId?: string;
  sentAt?: string;
  ackDeadlineAt?: string;
};

export interface GatewaySubmitDeadLetterDto {
  streamMessageId: string;
  traceId?: string;
  messageId?: string;
  channelId?: string;
  tenantId?: string;
  applicationId?: string;
  submitId?: string;
  failureCode: string;
  failureMessage: string;
  attempts: number;
  maxAttempts: number;
  commandPayload?: Record<string, unknown>;
  rawPayload?: string;
  deadLetteredAt?: string;
}

export interface GatewayDownstreamRecoveryStatusDto {
  account: string;
  gatewayInstanceId?: string;
  state: string;
  lockOwner?: string;
  lockExpiresAt?: string;
  lastAttemptAt?: string;
  lastSuccessAt?: string;
  lastFailureAt?: string;
  nextRetryAt?: string;
  attemptCount?: number;
  failureCategory?: string;
  lastError?: string;
  lastSkipReason?: string;
}

export interface TimeoutUnknownDto {
  olderThanHours?: number;
}

export interface ImportPreviewDto {
  tenantId: string;
  applicationId?: string;
  content: string;
  fileName?: string;
  encoding?: 'utf8' | 'gbk';
  delimiter?: ',' | '\t';
  requiredVariables?: string[];
}

export interface ConfirmImportDto extends CreateBatchTaskDto {
  importContent: string;
  requiredVariables?: string[];
}

interface SendJob {
  messageRecordId: string;
}

type QueuePriority = 'normal' | 'priority';

type RoutedChannel = {
  channel: {
    id: string;
    code: string;
    account: string;
    srcId: string;
    rateLimitPerSecond: number;
    unitPrice: number;
    status: string;
    carrier?: string | null;
    sendRegion: string;
    gatewayHost: string;
    gatewayPort: number;
    passwordCipher: string;
    cmppVersion: string;
    config?: unknown;
  };
  carrier: string;
  province?: string | null;
  groupId: string;
  routeScope: 'province' | 'national';
};

const SEND_QUEUE = 'sms.send.queue';
const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue';
const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands';
const DEFAULT_DOWNSTREAM_RETRY_DELAY_MS = 60_000;
const DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS = 30 * 60_000;
const DEFAULT_DOWNSTREAM_MAX_RETRIES = 10;
const BULLMQ_PRIORITY: Record<QueuePriority, number> = {
  priority: 1,
  normal: 100,
};

@Injectable()
export class SendChainService implements OnModuleInit, OnModuleDestroy {
  private redis?: IORedis;
  private sendQueue?: Queue<SendJob, unknown, 'send-message'>;
  private gatewayQueue?: Queue;
  private worker?: Worker<SendJob>;

  constructor(
    private readonly prisma: PrismaService,
    private readonly billing: BillingService,
    private readonly riskReview: RiskReviewService,
  ) {}

  onModuleInit() {
    if (process.env.API_ENABLE_SEND_WORKER === 'true') {
      this.startWorker();
    }
  }

  async onModuleDestroy() {
    await this.worker?.close();
    await this.sendQueue?.close();
    await this.gatewayQueue?.close();
    this.redis?.disconnect();
  }

  async createBatchTask(data: CreateBatchTaskDto) {
    const phones = [...new Set(data.phones ?? [])];
    const schedule = parseSchedule(data);
    await this.validateSendResources(data.tenantId, data.applicationId, data.templateId);
    const unitPrice = await this.resolveUnitPrice(data.tenantId, data.applicationId);
    const queuePriority = await this.resolveQueuePriority(data.tenantId, data.applicationId);
    const risk = await this.riskReview.evaluateTask({
      tenantId: data.tenantId,
      applicationId: data.applicationId,
      templateId: data.templateId,
      content: data.content,
      category: data.category,
      phones,
      variables: data.variables,
      createdById: data.createdById,
    });
    const billing = this.billing.estimateSmsCost({
      tenantId: data.tenantId,
      applicationId: data.applicationId,
      taskId: risk.task?.id,
      content: data.content,
      phoneCount: phones.length,
      unitPrice,
    });
    const batchStatus = statusFromRisk(risk.status, Boolean(schedule.scheduledAt));
    const shouldReserveBalance = batchStatus === 'ready';
    if (risk.status === 'approved') {
      const accountCheck = await this.billing.checkAccount({
        tenantId: data.tenantId,
        amountCents: billing.amountCents,
      });
      if (!accountCheck.canSend) {
        throw new BadRequestException('企业账户余额不足');
      }
    }
    const task = await this.prisma.smsBatchTask.create({
      data: {
        tenantId: data.tenantId,
        applicationId: data.applicationId,
        templateId: data.templateId,
        taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
        sourceType: data.sourceType ?? 'client',
        content: data.content,
        category: data.category,
        phoneTotal: phones.length,
        status: batchStatus,
        riskTaskId: risk.task?.id,
        auditStatus: risk.status === 'pending_review' ? 'pending' : risk.status === 'rejected' ? 'rejected' : 'approved',
        reviewReason: risk.status === 'pending_review' ? risk.reason : null,
        rejectReason: risk.status === 'rejected' ? risk.reason : null,
        progressTotal: phones.length,
        scheduledAt: schedule.scheduledAt,
        createdById: data.createdById,
      },
    });
    if (shouldReserveBalance && billing.amountCents > 0) {
      await this.billing.freeze({
        tenantId: data.tenantId,
        amountCents: billing.amountCents,
        relatedType: 'sms_batch_task',
        relatedId: task.id,
        remark: '发送任务创建冻结',
      });
    }
    await this.prisma.smsApiRequest.create({
      data: {
        tenantId: data.tenantId,
        batchTaskId: task.id,
        requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
        sourceIp: data.sourceIp,
        userAgent: data.userAgent,
        payloadSummary: {
          phoneTotal: phones.length,
          contentLength: [...data.content].length,
          category: data.category,
          sendMode: schedule.scheduledAt ? 'scheduled' : 'immediate',
          scheduledAt: schedule.scheduledAt?.toISOString(),
        },
        status: batchStatus === 'rejected' ? 'rejected' : 'accepted',
      },
    });
    if (phones.length > 0) {
      await this.prisma.smsMessageRecord.createMany({
        data: phones.map((phone) => ({
          tenantId: data.tenantId,
          batchTaskId: task.id,
          applicationId: data.applicationId,
          templateId: data.templateId,
          messageId: `MSG-${randomUUID()}`,
          phoneNumber: phone,
          content: data.content,
          billingUnits: billing.billingUnitsPerMessage,
          unitPrice: billing.unitPrice,
          amountCents: billing.billingUnitsPerMessage * billing.unitPrice,
          queuePriority,
          status: batchStatus === 'ready' ? 'queued' : batchStatus === 'scheduled' ? 'scheduled' : batchStatus,
          errorMessage: risk.status === 'rejected' ? risk.reason ?? undefined : undefined,
        })),
      });
    }
    if (batchStatus === 'ready') {
      await this.enqueueBatchTask(task.id);
    }
    return this.getBatchTask(task.id);
  }

  async listBatchTasks(tenantId?: string, status?: string, sourceType = 'client') {
    const tasks = await this.prisma.smsBatchTask.findMany({
      where: { tenantId, status, sourceType },
      include: {
        tenant: true,
        application: true,
        template: true,
        apiRequests: true,
      },
      orderBy: { createdAt: 'desc' },
    });
    const taskIds = tasks.map((task) => task.id);
    const messageStats = taskIds.length > 0 ? await this.prisma.smsMessageRecord.groupBy({
      by: ['batchTaskId', 'carrier', 'province', 'status'],
      where: { batchTaskId: { in: taskIds } },
      _count: { _all: true },
      _sum: { billingUnits: true },
    }) : [];
    return tasks.map((task) => ({
      ...task,
      messageStats: messageStats.filter((item) => item.batchTaskId === task.id),
    }));
  }

  async getBatchTask(taskId: string, tenantId?: string, sourceType = 'client') {
    const task = await this.prisma.smsBatchTask.findFirst({
      where: { id: taskId, tenantId, sourceType },
      include: { apiRequests: true, messages: { take: 20, orderBy: { queuedAt: 'asc' } } },
    });
    if (!task) {
      throw new NotFoundException('SMS batch task not found');
    }
    return task;
  }

  async listClientTaskMessages(taskId: string, tenantId: string) {
    await this.getBatchTask(taskId, tenantId, 'client');
    return this.listMessages({ tenantId, taskId });
  }

  listMessages(query: {
    tenantId?: string;
    applicationId?: string;
    channelId?: string;
    taskId?: string;
    phoneNumber?: string;
    status?: string;
  } = {}) {
    return this.prisma.smsMessageRecord.findMany({
      where: {
        tenantId: query.tenantId,
        applicationId: query.applicationId,
        channelId: query.channelId,
        batchTaskId: query.taskId,
        phoneNumber: query.phoneNumber,
        status: query.status,
      },
      include: {
        tenant: true,
        application: true,
        channel: true,
        submitRecords: { include: { channel: true }, orderBy: { createdAt: 'asc' } },
        receiptRecords: { include: { channel: true }, orderBy: { createdAt: 'asc' } },
      },
      orderBy: { queuedAt: 'desc' },
    });
  }

  listSubmitRecords(taskId?: string) {
    return this.prisma.smsSubmitRecord.findMany({
      where: { batchTaskId: taskId },
      orderBy: { createdAt: 'desc' },
    });
  }

  listReceiptRecords(taskId?: string) {
    return this.prisma.smsReceiptRecord.findMany({
      where: { batchTaskId: taskId },
      orderBy: { createdAt: 'desc' },
    });
  }

  listUplinkMessages(tenantId?: string, channelId?: string) {
    return this.prisma.smsUplinkMessage.findMany({
      where: { tenantId, channelId },
      include: { application: true, channel: true, messageRecord: { include: { application: true } } },
      orderBy: { receivedAt: 'desc' },
    });
  }

  async previewImport(data: ImportPreviewDto) {
    const sizeBytes = Buffer.byteLength(data.content, 'utf8');
    if (sizeBytes > 20 * 1024 * 1024) {
      throw new BadRequestException('导入文件不能超过 20MB');
    }
    const rows = parseImportRows(data.content, data.delimiter);
    const phones: string[] = [];
    const errors: Array<{ rowNumber: number; phoneNumber?: string; reason: string }> = [];
    const requiredVariables = data.requiredVariables ?? [];
    const enterpriseBlacklist = data.applicationId ? await this.prisma.enterpriseBlacklist.findMany({
      where: { tenantId: data.tenantId, applicationId: data.applicationId, status: 'active' },
      select: { phoneNumber: true },
    }) : [];
    const globalBlacklist = await this.prisma.globalBlacklist.findMany({
      where: { status: 'active' },
      select: { phoneNumber: true },
    });
    const blacklist = new Set([...enterpriseBlacklist, ...globalBlacklist].map((item) => item.phoneNumber));
    const seen = new Set<string>();
    for (const row of rows) {
      if (!row.phoneNumber) {
        errors.push({ rowNumber: row.rowNumber, reason: '缺少手机号' });
        continue;
      }
      if (!/^1[3-9]\d{9}$/.test(row.phoneNumber)) {
        errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '手机号格式非法' });
        continue;
      }
      if (seen.has(row.phoneNumber)) {
        errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '重复号码' });
        continue;
      }
      if (blacklist.has(row.phoneNumber)) {
        errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: '命中黑名单' });
        continue;
      }
      const missingVariables = requiredVariables.filter((name) => !row.variables[name]);
      if (missingVariables.length > 0) {
        errors.push({ rowNumber: row.rowNumber, phoneNumber: row.phoneNumber, reason: `变量列缺失：${missingVariables.join(',')}` });
        continue;
      }
      seen.add(row.phoneNumber);
      phones.push(row.phoneNumber);
    }
    return {
      fileName: data.fileName,
      encoding: data.encoding ?? 'utf8',
      totalRows: rows.length,
      validCount: phones.length,
      errorCount: errors.length,
      phones,
      errors,
    };
  }

  async confirmImport(data: ConfirmImportDto) {
    const preview = await this.previewImport({
      tenantId: data.tenantId,
      applicationId: data.applicationId,
      content: data.importContent,
      requiredVariables: data.requiredVariables,
    });
    if (preview.validCount === 0) {
      throw new BadRequestException('导入文件没有可发送号码');
    }
    return this.createBatchTask({ ...data, phones: preview.phones });
  }

  async enqueueBatchTask(taskId: string) {
    const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
    if (!task) {
      throw new NotFoundException('SMS batch task not found');
    }
    if (task.status === 'canceled') {
      throw new BadRequestException('SMS batch task is canceled');
    }
    const messages = await this.prisma.smsMessageRecord.findMany({
      where: { batchTaskId: taskId, status: 'queued' },
      select: { id: true, queuePriority: true },
      take: 100000,
    });
    const queue = this.getSendQueue();
    for (const message of messages) {
      const queuePriority = normalizeQueuePriority(message.queuePriority);
      await queue.add('send-message', { messageRecordId: message.id }, {
        jobId: message.id,
        attempts: 3,
        priority: BULLMQ_PRIORITY[queuePriority],
      });
    }
    await this.prisma.smsBatchTask.update({ where: { id: taskId }, data: { status: 'queued' } });
    return { taskId, enqueued: messages.length };
  }

  async cancelBatchTask(taskId: string, tenantId?: string, sourceType = 'client') {
    const task = await this.prisma.smsBatchTask.findFirst({ where: { id: taskId, tenantId, sourceType } });
    if (!task) {
      throw new NotFoundException('SMS batch task not found');
    }
    if (task.status !== 'scheduled') {
      throw new BadRequestException('Only scheduled SMS batch tasks can be canceled before dispatch');
    }
    await this.prisma.smsMessageRecord.updateMany({
      where: { batchTaskId: taskId, status: 'scheduled' },
      data: { status: 'canceled', errorMessage: '定时任务已取消' },
    });
    return this.prisma.smsBatchTask.update({
      where: { id: taskId },
      data: { status: 'canceled', canceledAt: new Date() },
    });
  }

  async handleReviewDecision(reviewTaskId: string, decision: 'approved' | 'rejected', reason: string) {
    const reviewTask = await this.prisma.smsSendTask.findUnique({
      where: { id: reviewTaskId },
      include: {
        messageRecords: {
          where: { status: 'pending_review' },
          include: { batchTask: true },
        },
      },
    });
    if (!reviewTask || reviewTask.messageRecords.length === 0) {
      return { reviewTaskId, decision, affected: 0 };
    }
    const batchTaskIds = new Set<string>();
    for (const message of reviewTask.messageRecords) {
      if (!message.tenantId || !message.applicationId || !message.batchTaskId) continue;
      if (decision === 'approved') {
        await this.prisma.smsMessageRecord.update({
          where: { id: message.id },
          data: { status: 'queued', errorCode: null, errorMessage: null },
        });
        await this.prisma.smsBatchTask.update({
          where: { id: message.batchTaskId },
          data: { status: 'ready', auditStatus: 'approved', reviewReason: reason, rejectReason: null },
        });
        batchTaskIds.add(message.batchTaskId);
      } else {
        await this.releaseMessageReservation(
          message as typeof message & { tenantId: string; batchTaskId: string },
          '模板不匹配人工审核驳回释放冻结',
        );
        await this.prisma.smsBatchTask.update({
          where: { id: message.batchTaskId },
          data: { status: 'rejected', auditStatus: 'rejected', rejectReason: reason },
        });
        await this.recordCmppFailureReceipt(message, 'REVIEW_REJECTED', reason);
      }
    }
    for (const batchTaskId of batchTaskIds) {
      await this.enqueueBatchTask(batchTaskId);
    }
    return { reviewTaskId, decision, affected: reviewTask.messageRecords.length };
  }

  async terminateBatchTask(taskId: string) {
    const task = await this.prisma.smsBatchTask.findUnique({ where: { id: taskId } });
    if (!task) {
      throw new NotFoundException('SMS batch task not found');
    }
    if (['finished', 'completed', 'failed', 'canceled', 'rejected'].includes(task.status)) {
      throw new BadRequestException('SMS batch task is already final');
    }
    await this.prisma.smsMessageRecord.updateMany({
      where: { batchTaskId: taskId, status: { in: ['ready', 'queued', 'scheduled', 'submit_queued'] } },
      data: { status: 'canceled', errorMessage: '运营终止任务，未提交号码停止发送' },
    });
    await this.refreshTaskProgress(taskId);
    return this.prisma.smsBatchTask.update({
      where: { id: taskId },
      data: { status: 'canceled', canceledAt: new Date(), rejectReason: '运营终止任务' },
    });
  }

  async dispatchDueScheduledTasks(now = new Date()) {
    const tasks = await this.prisma.smsBatchTask.findMany({
      where: { status: 'scheduled', scheduledAt: { lte: now } },
      orderBy: { scheduledAt: 'asc' },
    });
    const results: Array<{ taskId: string; status: string; enqueued?: number; reason?: string }> = [];
    for (const task of tasks) {
      try {
        await this.validateSendResources(task.tenantId, task.applicationId ?? undefined, task.templateId ?? undefined);
        const messages = await this.prisma.smsMessageRecord.findMany({
          where: { batchTaskId: task.id, status: 'scheduled' },
          select: { id: true, amountCents: true, billingUnits: true },
          take: 100000,
        });
        const amountCents = messages.reduce((sum, message) => sum + message.amountCents, 0);
        const accountCheck = await this.billing.checkAccount({ tenantId: task.tenantId, amountCents });
        if (!accountCheck.canSend) {
          throw new BadRequestException('定时任务到点时企业账户余额不足');
        }
        if (amountCents > 0) {
          await this.billing.freeze({
            tenantId: task.tenantId,
            amountCents,
            relatedType: 'sms_batch_task',
            relatedId: task.id,
            remark: '定时任务到点冻结',
          });
        }
        await this.prisma.smsMessageRecord.updateMany({
          where: { batchTaskId: task.id, status: 'scheduled' },
          data: { status: 'queued' },
        });
        const enqueued = await this.enqueueBatchTask(task.id);
        results.push({ taskId: task.id, status: 'queued', enqueued: enqueued.enqueued });
      } catch (error) {
        const reason = error instanceof Error ? error.message : '定时任务到点执行失败';
        await this.prisma.smsMessageRecord.updateMany({
          where: { batchTaskId: task.id, status: 'scheduled' },
          data: { status: 'rejected', errorMessage: reason },
        });
        await this.prisma.smsBatchTask.update({
          where: { id: task.id },
          data: { status: 'failed', rejectReason: reason },
        });
        results.push({ taskId: task.id, status: 'failed', reason });
      }
    }
    return { dispatched: results.filter((result) => result.status === 'queued').length, results };
  }

  startWorker() {
    if (this.worker) {
      return { status: 'already_started' };
    }
    const connection = bullmqConnection();
    this.worker = new Worker<SendJob>(
      SEND_QUEUE,
      async (job) => this.processSendJob(job.data),
      { connection, concurrency: Number(process.env.API_SEND_WORKER_CONCURRENCY ?? 20) },
    );
    return { status: 'started' };
  }

  async processSendJob(job: SendJob) {
    const message = await this.prisma.smsMessageRecord.findUnique({
      where: { id: job.messageRecordId },
      include: { batchTask: true, template: { include: { signature: true } }, signature: true },
    });
    if (!message || message.status !== 'queued') {
      return { skipped: true };
    }
    if (!message.tenantId || !message.batchTaskId) {
      return { skipped: true, reason: 'standalone channel test message' };
    }
    const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
    try {
      const routed = await this.selectChannelForMessage(businessMessage);
      return await this.submitMessageToGateway(businessMessage, routed, 0);
    } catch (error) {
      const reason = error instanceof Error ? error.message : '无可用通道组或通道';
      await this.prisma.smsMessageRecord.update({
        where: { id: message.id },
        data: { status: 'failed', errorMessage: reason },
      });
      await this.releaseMessageReservation(businessMessage, reason);
      if (message.batchTask?.sourceType === 'cmpp') {
        await this.recordCmppFailureReceipt(businessMessage, 'ROUTE', reason);
      } else {
        await this.refreshTaskProgress(businessMessage.batchTaskId);
      }
      return { submitted: false, messageRecordId: message.id, status: 'failed', reason };
    }
  }

  async handleSubmitResult(data: GatewaySubmitResultDto) {
    const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
    const batchTask = message.batchTaskId
      ? await this.prisma.smsBatchTask.findUnique({ where: { id: message.batchTaskId }, select: { sourceType: true } })
      : null;
    const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
    await this.prisma.smsSubmitRecord.updateMany({
      where: { OR: [{ submitId: data.submitId }, { messageRecordId: message.id }] },
      data: {
        sequenceId: data.sequenceId,
        gatewayMessageId: data.gatewayMessageId,
        submitStatus: data.submitStatus,
        errorCode: data.errorCode,
        errorMessage: data.errorMessage,
        submittedAt,
      },
    });
    await this.recordSubmitSegments(message, data, submittedAt);
    const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId;
    if (data.submitId && message.submitId && data.submitId !== message.submitId) {
      return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
    }
    const status = data.submitStatus === 'accepted' ? 'submitted' : data.submitStatus === 'timeout' ? 'timeout' : 'submit_failed';
    if (data.submitStatus === 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
      const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
      await this.chargeAcceptedMessage(businessMessage);
    } else if (data.submitStatus !== 'accepted' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
      const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
      const retried = await this.retryMessageIfAllowed(businessMessage, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发');
      if (retried) {
        await this.refreshTaskProgress(businessMessage.batchTaskId);
        return retried;
      }
      await this.releaseMessageReservation(businessMessage, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结');
    }
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: {
        gatewayMessageId: data.gatewayMessageId,
        submitStatus: data.submitStatus,
        status,
        errorCode: data.errorCode,
        errorMessage: data.errorMessage,
        submittedAt,
        timeoutAt: data.submitStatus === 'timeout' ? submittedAt : undefined,
      },
    });
    if (data.submitStatus !== 'accepted' && batchTask?.sourceType === 'cmpp' && message.tenantId && message.applicationId) {
      await this.recordCmppFailureReceipt(
        message as typeof message & { tenantId: string; applicationId: string; batchTaskId: string },
        data.errorCode || 'SUBMIT',
        data.errorMessage || (data.submitStatus === 'timeout' ? '上游提交超时' : '上游拒绝短信'),
      );
    }
    await this.prisma.gatewaySubmitDeadLetter.updateMany({
      where: {
        status: { in: ['pending', 'requeued'] },
        OR: [
          data.submitId ? { submitId: data.submitId } : undefined,
          data.messageId ? { messageId: data.messageId } : undefined,
        ].filter(Boolean) as Array<{ submitId?: string; messageId?: string }>,
      },
      data: {
        status: 'resolved',
        resolvedAt: submittedAt,
        resolvedStatus: data.submitStatus,
      },
    });
    if (message.batchTaskId) {
      await this.refreshTaskProgress(message.batchTaskId);
    }
    return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
  }

  async handleReceipt(data: GatewayReceiptEventDto) {
    const resolved = await this.resolveReceiptMessage(data);
    const message = resolved.message;
    const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
    const status =
      data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed';
    if (resolved.submitRecordId) {
      await this.prisma.smsSubmitRecord.updateMany({
        where: {
          id: resolved.submitRecordId,
          gatewayMessageId: null,
        },
        data: {
          gatewayMessageId: data.gatewayMessageId,
          sequenceId: data.sequenceId,
        },
      });
    }
    await this.prisma.smsReceiptRecord.create({
      data: {
        tenantId: message.tenantId,
        batchTaskId: message.batchTaskId,
        messageRecordId: message.id,
        channelId: data.channelId,
        messageId: resolved.messageId,
        gatewayMessageId: data.gatewayMessageId,
        sequenceId: data.sequenceId,
        receiptStatus: data.receiptStatus,
        rawStatus: data.rawStatus,
        errorCode: data.errorCode,
        deliveredAt,
      },
    });
    await this.recordReceiptSegment(message, data, deliveredAt, resolved.submitRecordId);
    const isCurrentAttempt =
      (!message.channelId || message.channelId === data.channelId)
      && (!message.gatewayMessageId || message.gatewayMessageId === data.gatewayMessageId);
    if (!isCurrentAttempt || (status === 'failed' && message.status === 'delivered')) {
      return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
    }
    const isStandaloneChannelTest = !message.tenantId && !message.batchTaskId;
    if (status === 'failed' && !isStandaloneChannelTest && message.tenantId && message.batchTaskId) {
      const businessMessage = message as typeof message & { tenantId: string; batchTaskId: string };
      const retried = await this.retryMessageIfAllowed(businessMessage, '回执失败补发');
      if (retried) {
        await this.refreshTaskProgress(businessMessage.batchTaskId);
        return retried;
      }
      await this.refundMessage(businessMessage, '最终失败退款');
    }
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: {
        receiptStatus: data.receiptStatus,
        status,
        errorCode: data.errorCode,
        deliveredAt,
      },
    });
    if (!isStandaloneChannelTest && message.tenantId && message.applicationId) {
      await this.queueAndTryDownstreamDelivery({
        tenantId: message.tenantId,
        applicationId: message.applicationId,
        messageRecordId: message.id,
        messageId: message.messageId,
        deliveryType: 'receipt',
        payload: {
          messageId: message.messageId,
          gatewayMessageId: data.gatewayMessageId,
          phoneNumber: message.phoneNumber,
          receiptStatus: data.receiptStatus,
          rawStatus: data.rawStatus,
          errorCode: data.errorCode,
          deliveredAt: deliveredAt.toISOString(),
        },
      });
    }
    if (message.batchTaskId) {
      await this.refreshTaskProgress(message.batchTaskId);
    }
    return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
  }

  async handleUplink(data: GatewayUplinkEventDto) {
    const channel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
    if (!channel) {
      throw new NotFoundException('SMS channel not found');
    }
    const match = await this.resolveUplinkMatch(data, channel);
    const record = await this.prisma.smsUplinkMessage.create({
      data: {
        tenantId: match.tenantId,
        applicationId: match.applicationId,
        messageRecordId: match.messageRecordId,
        channelId: data.channelId,
        messageId: data.messageId,
        sequenceId: 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.queueAndTryDownstreamDelivery({
        tenantId: match.tenantId,
        applicationId: match.applicationId,
        messageRecordId: match.messageRecordId,
        messageId: data.messageId,
        deliveryType: 'uplink',
        payload: {
          messageId: data.messageId,
          applicationId: match.applicationId,
          phoneNumber: data.phoneNumber,
          destId: data.destId,
          content: data.content,
          receivedAt: record.receivedAt.toISOString(),
        },
      });
    }
    return record;
  }

  async listPendingDownstreamDeliveries(data: GatewayPendingDeliveryQueryDto) {
    const application = await this.findInboundApplication(data.account);
    if (!application || application.status !== 'active' || application.tenant.status !== 'active') {
      throw new BadRequestException('CMPP account is invalid or disabled');
    }
    const expiredAcknowledgements = await this.prisma.cmppDownstreamDelivery.findMany({
      where: { applicationId: application.id, status: 'awaiting_ack', ackDeadlineAt: { lte: new Date() } },
      select: { id: true },
      take: 500,
    });
    for (const expired of expiredAcknowledgements) {
      await this.markDownstreamDeliveryFailed(expired.id, 'CMPP_DELIVER_RESP timeout recovered after Gateway restart', 'ack_timeout');
    }
    return this.prisma.cmppDownstreamDelivery.findMany({
      where: {
        applicationId: application.id,
        status: 'pending',
        OR: [{ nextRetryAt: null }, { nextRetryAt: { lte: new Date() } }],
      },
      orderBy: { createdAt: 'asc' },
      take: Math.min(Math.max(data.limit ?? 100, 1), 500),
    });
  }

  async markDownstreamDeliveryDelivered(id: string) {
    return this.prisma.cmppDownstreamDelivery.update({
      where: { id },
      data: {
        status: 'delivered',
        deliveredAt: new Date(),
        lastError: null,
      },
    });
  }

  async markDownstreamDeliverySent(data: GatewayDownstreamSentDto) {
    const sentAt = asDateOrNull(data.sentAt) ?? new Date();
    const ackDeadlineAt = asDateOrNull(data.ackDeadlineAt) ?? new Date(sentAt.getTime() + downstreamAckTimeoutMs());
    await this.prisma.cmppDownstreamDelivery.updateMany({
      where: { id: data.id, status: { not: 'delivered' } },
      data: {
        status: 'awaiting_ack',
        sentAt,
        ackDeadlineAt,
        ackSequenceId: data.sequenceId,
        ackMessageId: data.messageId,
        connectionId: data.connectionId,
        nextRetryAt: null,
        lastError: null,
      },
    });
    return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } });
  }

  async acknowledgeDownstreamDelivery(data: GatewayDownstreamAcknowledgedDto) {
    const acknowledgedAt = asDateOrNull(data.acknowledgedAt) ?? new Date();
    const acknowledgedMessageId = String(data.messageId ?? '').trim();
    if (data.result === 0 && acknowledgedMessageId !== '' && acknowledgedMessageId !== '0') {
      await this.prisma.cmppDownstreamDelivery.updateMany({
        where: { id: data.id, status: { not: 'delivered' } },
        data: {
          status: 'delivered',
          acknowledgedAt,
          deliveredAt: acknowledgedAt,
          ackDeadlineAt: null,
          ackResult: data.result,
          ackSequenceId: data.sequenceId,
          ackMessageId: data.messageId,
          connectionId: data.connectionId,
          nextRetryAt: null,
          lastError: null,
        },
      });
      return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: data.id } });
    }
    await this.prisma.cmppDownstreamDelivery.updateMany({
      where: { id: data.id, status: { not: 'delivered' } },
      data: {
        acknowledgedAt,
        ackResult: data.result,
        ackSequenceId: data.sequenceId,
        ackMessageId: data.messageId,
        connectionId: data.connectionId,
      },
    });
    if (data.result === 0) {
      return this.markDownstreamDeliveryFailed(data.id, 'CMPP_DELIVER_RESP Msg_Id=0，客户端仅确认协议收包，无法关联原短信', 'ack_invalid');
    }
    return this.markDownstreamDeliveryFailed(data.id, `downstream CMPP_DELIVER_RESP result=${data.result}`, 'ack_rejected');
  }

  async markDownstreamDeliveryFailed(id: string, errorMessage?: string, failureType: GatewayDownstreamFailureType = 'send_failed') {
    const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({ where: { id } });
    if (!delivery) {
      throw new NotFoundException('Downstream delivery not found');
    }
    if (delivery.status === 'delivered') {
      return delivery;
    }
    const retryCount = (delivery.retryCount ?? 0) + 1;
    const acknowledgementFailure = failureType === 'ack_timeout' || failureType === 'ack_rejected' || failureType === 'ack_invalid' || failureType === 'connection_lost';
    const retryAllowed = !acknowledgementFailure || delivery.retryEnabled !== false;
    const finalFailure = !retryAllowed || retryCount >= downstreamMaxRetries();
    const finalStatus = failureType === 'ack_rejected' ? 'rejected' : acknowledgementFailure ? 'unconfirmed' : 'failed';
    const updated = await this.prisma.cmppDownstreamDelivery.update({
      where: { id },
      data: {
        status: finalFailure ? finalStatus : 'pending',
        retryCount,
        nextRetryAt: finalFailure ? null : new Date(Date.now() + downstreamRetryDelayMs(retryCount)),
        ackDeadlineAt: null,
        lastError: errorMessage ?? 'downstream delivery failed',
      },
    });
    if (finalFailure) {
      await this.prisma.operationLog.create({
        data: {
          tenantId: updated.tenantId,
          action: 'gateway.downstream_delivery_failed',
          resource: 'cmpp_downstream_delivery',
          resourceId: updated.id,
          detail: {
            deliveryType: updated.deliveryType,
            applicationId: updated.applicationId,
            messageId: updated.messageId,
            retryCount,
            failureType,
            retryEnabled: updated.retryEnabled,
            errorMessage: updated.lastError,
          },
        },
      });
    }
    return updated;
  }

  async recordGatewaySubmitDeadLetter(data: GatewaySubmitDeadLetterDto) {
    const createdAt = data.deadLetteredAt ? new Date(data.deadLetteredAt) : new Date();
    return this.prisma.gatewaySubmitDeadLetter.upsert({
      where: { streamMessageId: data.streamMessageId },
      update: {
        tenantId: data.tenantId,
        applicationId: data.applicationId,
        channelId: data.channelId,
        traceId: data.traceId,
        messageId: data.messageId,
        submitId: data.submitId,
        status: 'pending',
        failureCode: data.failureCode,
        failureMessage: data.failureMessage,
        attempts: data.attempts,
        maxAttempts: data.maxAttempts,
        commandPayload: data.commandPayload as Prisma.InputJsonValue | undefined,
        rawPayload: data.rawPayload,
        resolvedAt: null,
        resolvedStatus: null,
      },
      create: {
        streamMessageId: data.streamMessageId,
        tenantId: data.tenantId,
        applicationId: data.applicationId,
        channelId: data.channelId,
        traceId: data.traceId,
        messageId: data.messageId,
        submitId: data.submitId,
        failureCode: data.failureCode,
        failureMessage: data.failureMessage,
        attempts: data.attempts,
        maxAttempts: data.maxAttempts,
        commandPayload: data.commandPayload as Prisma.InputJsonValue | undefined,
        rawPayload: data.rawPayload,
        createdAt,
      },
    });
  }

  async recordGatewayDownstreamRecoveryStatus(data: GatewayDownstreamRecoveryStatusDto) {
    const account = String(data.account ?? '').trim();
    if (!account) {
      throw new BadRequestException('account is required');
    }
    const recoveryStatuses = (this.prisma as PrismaService & {
      gatewayDownstreamRecoveryStatus: {
        findUnique: (args: Record<string, unknown>) => Promise<any>;
        upsert: (args: Record<string, unknown>) => Promise<any>;
      };
    }).gatewayDownstreamRecoveryStatus;
    const previous = await recoveryStatuses.findUnique({
      where: { account },
      select: {
        state: true,
        gatewayInstanceId: true,
        lockOwner: true,
        failureCategory: true,
        lastError: true,
        lastSkipReason: true,
      },
    });
    const application = await this.prisma.smsApplication.findUnique({
      where: { cmppAccount: account },
      select: { id: true, tenantId: true, name: true },
    });
    const failureCategory = normalizeRecoveryFailureCategory(data);
    const updated = await recoveryStatuses.upsert({
      where: { account },
      update: {
        tenantId: application?.tenantId ?? null,
        applicationId: application?.id ?? null,
        gatewayInstanceId: data.gatewayInstanceId ?? null,
        state: data.state,
        lockOwner: data.lockOwner ?? null,
        lockExpiresAt: asDateOrNull(data.lockExpiresAt),
        lastAttemptAt: asDateOrNull(data.lastAttemptAt),
        lastSuccessAt: asDateOrNull(data.lastSuccessAt),
        lastFailureAt: asDateOrNull(data.lastFailureAt),
        nextRetryAt: asDateOrNull(data.nextRetryAt),
        attemptCount: Number.isFinite(Number(data.attemptCount)) ? Number(data.attemptCount) : 0,
        failureCategory,
        lastError: data.lastError ?? null,
        lastSkipReason: data.lastSkipReason ?? null,
      },
      create: {
        account,
        tenantId: application?.tenantId,
        applicationId: application?.id,
        gatewayInstanceId: data.gatewayInstanceId,
        state: data.state,
        lockOwner: data.lockOwner,
        lockExpiresAt: asDateOrNull(data.lockExpiresAt),
        lastAttemptAt: asDateOrNull(data.lastAttemptAt),
        lastSuccessAt: asDateOrNull(data.lastSuccessAt),
        lastFailureAt: asDateOrNull(data.lastFailureAt),
        nextRetryAt: asDateOrNull(data.nextRetryAt),
        attemptCount: Number.isFinite(Number(data.attemptCount)) ? Number(data.attemptCount) : 0,
        failureCategory,
        lastError: data.lastError,
        lastSkipReason: data.lastSkipReason,
      },
      include: {
        tenant: true,
        application: true,
      },
    });
    const normalizedUpdated = updated as typeof updated & {
      failureCategory?: string | null;
      lockOwner?: string | null;
      lockExpiresAt?: Date | null;
    };
    if (hasRecoveryAuditStateChanged(previous, updated)) {
      await this.prisma.operationLog.create({
        data: {
          tenantId: updated.tenantId ?? undefined,
          action: 'gateway.downstream_recovery_status_changed',
          resource: 'gateway_downstream_recovery_status',
          resourceId: updated.id,
          detail: {
            account,
            previousState: previous?.state ?? null,
            state: updated.state,
            gatewayInstanceId: updated.gatewayInstanceId,
            lockOwner: normalizedUpdated.lockOwner,
            attemptCount: updated.attemptCount,
            nextRetryAt: updated.nextRetryAt,
            failureCategory: normalizedUpdated.failureCategory,
            applicationId: updated.applicationId,
            applicationName: application?.name,
            lastError: updated.lastError,
            lastSkipReason: updated.lastSkipReason,
          },
        },
      });
    }
    return updated;
  }

  async requeueGatewaySubmitDeadLetter(id: string) {
    const deadLetter = await this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } });
    if (!deadLetter) {
      throw new NotFoundException('Gateway submit dead letter not found');
    }
    if (!deadLetter.commandPayload || typeof deadLetter.commandPayload !== 'object') {
      throw new BadRequestException('该死信缺少可重放的 SubmitCommand');
    }
    const retryStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload);
    const updated = await this.prisma.gatewaySubmitDeadLetter.update({
      where: { id },
      data: {
        status: 'requeued',
        manualRetryCount: { increment: 1 },
        lastRetryStreamId: retryStreamMessageId,
        lastRetriedAt: new Date(),
      },
    });
    await this.prisma.operationLog.create({
      data: {
        tenantId: updated.tenantId ?? undefined,
        action: 'gateway.submit_dead_letter_requeue',
        resource: 'gateway_submit_dead_letter',
        resourceId: updated.id,
        detail: {
          streamMessageId: updated.streamMessageId,
          retryStreamMessageId,
          submitId: updated.submitId,
          messageId: updated.messageId,
        },
      },
    });
    return updated;
  }

  async requeueDownstreamDelivery(id: string) {
    const delivery = await this.prisma.cmppDownstreamDelivery.findUnique({
      where: { id },
      include: { application: { select: { cmppAccount: true } } },
    });
    if (!delivery) {
      throw new NotFoundException('Downstream delivery not found');
    }
    const payload = isObjectRecord(delivery.payload) ? { ...delivery.payload } : null;
    if (!payload) {
      throw new BadRequestException('下游投递记录缺少可重放 payload');
    }
    const path =
      delivery.deliveryType === 'receipt'
        ? '/downstream/receipt'
        : delivery.deliveryType === 'uplink'
          ? '/downstream/uplink'
          : null;
    if (!path) {
      throw new BadRequestException(`Unsupported downstream delivery type ${delivery.deliveryType}`);
    }

    await this.prisma.operationLog.create({
      data: {
        tenantId: delivery.tenantId,
        action: 'gateway.downstream_delivery_requeue',
        resource: 'cmpp_downstream_delivery',
        resourceId: delivery.id,
        detail: {
          deliveryType: delivery.deliveryType,
          applicationId: delivery.applicationId,
          messageId: delivery.messageId,
        },
      },
    });

    const requestPayload = {
      deliveryId: delivery.id,
      account: String(payload.account ?? delivery.application?.cmppAccount ?? ''),
      ...payload,
    };
    await this.prisma.cmppDownstreamDelivery.update({
      where: { id: delivery.id },
      data: {
        status: 'pending',
        retryCount: 0,
        nextRetryAt: null,
        sentAt: null,
        acknowledgedAt: null,
        ackDeadlineAt: null,
        ackResult: null,
        ackSequenceId: null,
        ackMessageId: null,
        connectionId: null,
        deliveredAt: null,
        lastError: null,
      },
    });
    try {
      const result = await this.postGatewayControl(path, requestPayload) as GatewayControlDeliveryResult;
      if (result.sent || result.delivered) {
        return this.markDownstreamDeliverySent({ id: delivery.id, ...result });
      }
      return this.prisma.cmppDownstreamDelivery.findUnique({ where: { id: delivery.id } });
    } catch (error) {
      return this.markDownstreamDeliveryFailed(
        delivery.id,
        error instanceof Error ? error.message : 'Gateway control delivery failed',
      );
    }
  }

  async batchRequeueDownstreamDeliveries(ids: string[]) {
    const uniqueIds = [...new Set(ids.filter(Boolean))];
    if (uniqueIds.length === 0) {
      throw new BadRequestException('请选择至少一条下游投递记录');
    }
    const results: Array<{ id: string; status: 'success' | 'failed'; errorMessage?: string }> = [];
    for (const id of uniqueIds) {
      try {
        await this.requeueDownstreamDelivery(id);
        results.push({ id, status: 'success' });
      } catch (error) {
        results.push({
          id,
          status: 'failed',
          errorMessage: error instanceof Error ? error.message : '批量重投失败',
        });
      }
    }
    return {
      total: uniqueIds.length,
      successCount: results.filter((item) => item.status === 'success').length,
      failedCount: results.filter((item) => item.status === 'failed').length,
      results,
    };
  }

  async claimUplinkMatchCandidate(uplinkMessageId: string, candidateId: string, operatorId?: string) {
    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('该上行记录已完成匹配，不能重复认领');
    }
    const claimedAt = new Date();
    const messageId = candidate.uplinkMessage.messageId ?? candidate.messageRecord?.messageId ?? null;
    const [updatedUplink] = await this.prisma.$transaction([
      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}`,
        },
      }),
      this.prisma.smsUplinkMatchCandidate.updateMany({
        where: {
          uplinkMessageId,
          id: { not: candidate.id },
          status: 'pending',
        },
        data: { status: 'rejected' },
      }),
      this.prisma.smsUplinkMatchCandidate.update({
        where: { id: candidate.id },
        data: {
          status: 'claimed',
          claimedAt,
          claimedById: operatorId,
        },
      }),
      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.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' }],
        },
      },
    });
  }

  private async queueAndTryDownstreamDelivery(data: {
    tenantId: string;
    applicationId?: string | null;
    messageRecordId?: string | null;
    messageId?: string | null;
    deliveryType: 'receipt' | 'uplink';
    payload: Record<string, unknown>;
  }) {
    if (!data.applicationId) {
      return null;
    }
    const application = await this.prisma.smsApplication.findUnique({
      where: { id: data.applicationId },
      select: { cmppAccount: true, downstreamReceiptRetryEnabled: true, downstreamUplinkRetryEnabled: true },
    });
    const payload = { account: application?.cmppAccount, applicationId: data.applicationId, ...data.payload };
    const delivery = await this.prisma.cmppDownstreamDelivery.create({
      data: {
        tenantId: data.tenantId,
        applicationId: data.applicationId,
        messageRecordId: data.messageRecordId,
        messageId: data.messageId,
        deliveryType: data.deliveryType,
        payload,
        retryEnabled: data.deliveryType === 'uplink'
          ? application?.downstreamUplinkRetryEnabled ?? true
          : application?.downstreamReceiptRetryEnabled ?? true,
        status: 'pending',
      },
    });
    try {
      const result = await this.postGatewayControl(
        data.deliveryType === 'receipt' ? '/downstream/receipt' : '/downstream/uplink',
        { deliveryId: delivery.id, ...payload },
      ) as GatewayControlDeliveryResult;
      if (result.sent || result.delivered) {
        await this.markDownstreamDeliverySent({ id: delivery.id, ...result });
      }
    } catch (error) {
      await this.markDownstreamDeliveryFailed(delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed');
    }
    return delivery;
  }

  private async resolveUplinkMatch(
    data: GatewayUplinkEventDto,
    channel: { id: string; srcId?: string | null },
  ): Promise<{
    tenantId?: string;
    applicationId?: string;
    messageRecordId?: string;
    matchStatus: string;
    matchReason: string;
    candidates: UplinkMatchCandidateInput[];
  }> {
    if (data.messageId) {
      const message = await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } });
      if (message?.tenantId) {
        return {
          tenantId: message.tenantId,
          applicationId: message.applicationId ?? undefined,
          messageRecordId: message.id,
          matchStatus: message.applicationId ? 'matched' : 'unmatched',
          matchReason: message.applicationId ? 'messageId 精确匹配' : 'messageId 匹配到下发记录但无应用',
          candidates: [],
        };
      }
    }

    const accessNumber = data.destId || channel.srcId || '';
    const accessRoutes = accessNumber
      ? await this.prisma.channelRouteRule.findMany({
        where: {
          applicationId: { not: null },
          status: 'active',
          group: { items: { some: { channelId: channel.id, channel: { srcId: accessNumber } } } },
        },
        select: { applicationId: true },
        take: 10,
      })
      : [];
    const accessApplicationIds = [...new Set(accessRoutes.map((route) => route.applicationId).filter((value): value is string => Boolean(value)))];
    const accessApplications = accessApplicationIds.length > 0
      ? await this.prisma.smsApplication.findMany({
        where: { id: { in: accessApplicationIds }, status: 'active' },
        select: { id: true, tenantId: true, name: true },
      })
      : [];
    if (accessApplications.length === 1) {
      return {
        tenantId: accessApplications[0].tenantId,
        applicationId: accessApplications[0].id,
        matchStatus: 'matched',
        matchReason: '接入号唯一匹配应用',
        candidates: [],
      };
    }
    if (accessApplications.length > 1) {
      return {
        matchStatus: 'ambiguous',
        matchReason: '接入号匹配多个应用',
        candidates: accessApplications.map((application) => ({
          tenantId: application.tenantId,
          applicationId: application.id,
          matchSource: 'access_number',
          confidence: 70,
          reason: `接入号 ${accessNumber} 可匹配应用 ${application.name}`,
        })),
      };
    }

    const windowHours = Number(process.env.UPLINK_MATCH_WINDOW_HOURS ?? 72);
    const since = new Date(Date.now() - Math.max(1, windowHours) * 60 * 60 * 1000);
    const recentMessages = await this.prisma.smsMessageRecord.findMany({
      where: {
        phoneNumber: data.phoneNumber,
        tenantId: { not: null },
        applicationId: { not: null },
        submittedAt: { gte: since },
      },
      orderBy: { submittedAt: 'desc' },
      take: 2,
    });
    const matchableRecentMessages = recentMessages.filter((message) => message.tenantId && message.applicationId);
    if (matchableRecentMessages.length === 1) {
      return {
        tenantId: matchableRecentMessages[0].tenantId ?? undefined,
        applicationId: matchableRecentMessages[0].applicationId ?? undefined,
        messageRecordId: matchableRecentMessages[0].id,
        matchStatus: 'matched',
        matchReason: `手机号 ${windowHours} 小时窗口唯一匹配`,
        candidates: [],
      };
    }
    if (matchableRecentMessages.length > 1) {
      return {
        matchStatus: 'ambiguous',
        matchReason: `手机号 ${windowHours} 小时窗口匹配多条下发记录`,
        candidates: matchableRecentMessages
          .map((message) => ({
            tenantId: String(message.tenantId),
            applicationId: String(message.applicationId),
            messageRecordId: message.id,
            matchSource: 'phone_window',
            confidence: 55,
            reason: `手机号 ${windowHours} 小时窗口候选下发 ${message.messageId}`,
          })),
      };
    }
    return { matchStatus: 'unmatched', matchReason: '未匹配到应用或下发记录', candidates: [] };
  }

  async authenticateInboundApplication(data: GatewayInboundAuthDto) {
    const application = await this.findInboundApplication(data.account);
    if (!application || application.status !== 'active' || application.tenant.status !== 'active') {
      throw new BadRequestException('CMPP account is invalid or disabled');
    }
    if (!application.interfaceEnabled) {
      throw new BadRequestException('CMPP interface is disabled for this application');
    }
    if (application.tenant.certificationStatus !== 'approved') {
      throw new BadRequestException('Enterprise certification is not approved');
    }
    if (!matchesApplicationSecret(data, application.secretHash)) {
      throw new BadRequestException('CMPP account or password is invalid');
    }
    if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
      throw new BadRequestException('CMPP source IP is not in application allowlist');
    }
    return {
      applicationId: application.id,
      tenantId: application.tenantId,
      account: application.cmppAccount,
      enterpriseCode: application.cmppEnterpriseCode,
      passwordCipher: application.secretHash,
      status: 'authenticated',
    };
  }

  async submitInboundMessage(data: GatewayInboundSubmitDto) {
    const application = await this.findInboundApplication(data.account);
    if (!application) {
      throw new BadRequestException('CMPP account is invalid');
    }
    if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
      throw new BadRequestException('CMPP source IP is not in application allowlist');
    }
    if (!/^1[3-9]\d{9}$/.test(data.phoneNumber)) {
      throw new BadRequestException('CMPP submit phone number is invalid');
    }
    const template = await this.resolveInboundTemplateCandidate(application.id, data.content);
    const unitPrice = application.customerUnitPrice ?? 0;
    const queuePriority = normalizeQueuePriority(application.queuePriority);
    const billing = this.billing.estimateSmsCost({
      tenantId: application.tenantId,
      applicationId: application.id,
      content: data.content,
      phoneCount: 1,
      unitPrice,
    });
    const task = await this.prisma.smsBatchTask.create({
      data: {
        tenantId: application.tenantId,
        applicationId: application.id,
        templateId: template?.id,
        taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
        sourceType: 'cmpp',
        content: data.content,
        phoneTotal: 1,
        status: 'validating',
        progressTotal: 1,
      },
    });
    await this.prisma.smsApiRequest.create({
      data: {
        tenantId: application.tenantId,
        batchTaskId: task.id,
        requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
        sourceIp: data.remoteIp,
        userAgent: 'cmpp-gateway',
        payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account },
        status: 'accepted',
      },
    });
    const message = await this.prisma.smsMessageRecord.create({
      data: {
        tenantId: application.tenantId,
        batchTaskId: task.id,
        applicationId: application.id,
        templateId: template?.id,
        messageId: `MSG-${randomUUID()}`,
        phoneNumber: data.phoneNumber,
        content: data.content,
        billingUnits: billing.billingUnitsPerMessage,
        unitPrice: billing.unitPrice,
        amountCents: billing.amountCents,
        queuePriority,
        cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId),
        status: 'validating',
      },
    });

    const reject = async (code: string, reason: string) => {
      await this.prisma.smsBatchTask.update({
        where: { id: task.id },
        data: { status: 'rejected', auditStatus: 'rejected', rejectReason: reason },
      });
      await this.recordCmppFailureReceipt(message, code, reason);
    };
    if (application.status !== 'active' || application.tenant.status !== 'active') {
      await reject('ACCOUNT', '企业或短信应用已停用');
    } else if (!application.interfaceEnabled) {
      await reject('INTERFACE', '短信应用 CMPP 接口已停用');
    } else if (application.tenant.certificationStatus !== 'approved') {
      await reject('CERT', '企业认证未通过');
    } else if (!template && application.templateMismatchMode === 'manual_review') {
      const signature = await this.resolveInboundSignatureCandidate(application.id, data.content);
      if (!signature) {
        await reject('SIGNATURE', '短信内容未识别到已审核通过的签名');
      } else {
        const risk = await this.riskReview.evaluateTask({
          tenantId: application.tenantId,
          applicationId: application.id,
          content: data.content,
          phones: [data.phoneNumber],
        });
        if (risk.status === 'rejected') {
          await reject('RISK', risk.reason || '短信被风控拒绝');
        } else {
          const accountCheck = await this.billing.checkAccount({
            tenantId: application.tenantId,
            amountCents: billing.amountCents,
          });
          if (!accountCheck.canSend) {
            await reject('BALANCE', '企业账户余额不足');
          } else {
            if (billing.amountCents > 0) {
              await this.billing.freeze({
                tenantId: application.tenantId,
                amountCents: billing.amountCents,
                relatedType: 'sms_batch_task',
                relatedId: task.id,
                remark: 'CMPP 模板不匹配待审核短信冻结',
              });
            }
            const reviewTask = risk.status === 'pending_review' && risk.task
              ? await this.attachMessageToReviewTask(risk.task.id, message.id, signature.id)
              : await this.riskReview.aggregateTemplateMismatch({
                tenantId: application.tenantId,
                applicationId: application.id,
                account: data.account,
                messageRecordId: message.id,
                signatureId: signature.id,
                content: data.content,
              });
            await this.prisma.smsBatchTask.update({
              where: { id: task.id },
              data: {
                status: 'pending_review',
                riskTaskId: reviewTask?.id,
                auditStatus: 'pending',
                reviewReason: reviewTask?.reviewReason ?? '模板不匹配，等待人工审核',
              },
            });
          }
        }
      }
    } else if (!template) {
      await reject('TEMPLATE', '短信内容未匹配到已报备模板');
    } else if (template.auditStatus !== 'approved') {
      await reject('TEMPLATE', '短信模板尚未审核通过');
    } else if (!template.signature || template.signature.auditStatus !== 'approved') {
      await reject('SIGNATURE', '短信签名尚未审核通过');
    } else {
      const risk = await this.riskReview.evaluateTask({
        tenantId: application.tenantId,
        applicationId: application.id,
        templateId: template.id,
        content: data.content,
        phones: [data.phoneNumber],
      });
      if (risk.status === 'rejected') {
        await reject('RISK', risk.reason || '短信被风控拒绝');
      } else if (risk.status === 'pending_review') {
        await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'pending_review' } });
        await this.prisma.smsBatchTask.update({
          where: { id: task.id },
          data: { status: 'pending_review', riskTaskId: risk.task?.id, auditStatus: 'pending', reviewReason: risk.reason },
        });
      } else {
        const accountCheck = await this.billing.checkAccount({
          tenantId: application.tenantId,
          amountCents: billing.amountCents,
        });
        if (!accountCheck.canSend) {
          await reject('BALANCE', '企业账户余额不足');
        } else {
          if (billing.amountCents > 0) {
            await this.billing.freeze({
              tenantId: application.tenantId,
              amountCents: billing.amountCents,
              relatedType: 'sms_batch_task',
              relatedId: task.id,
              remark: 'CMPP 入站短信冻结',
            });
          }
          await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'queued' } });
          await this.prisma.smsBatchTask.update({
            where: { id: task.id },
            data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' },
          });
          await this.enqueueBatchTask(task.id);
        }
      }
    }
    return {
      accepted: true,
      tenantId: application.tenantId,
      applicationId: application.id,
      taskId: task.id,
      messageId: message.messageId,
      messageRecordId: message.id,
      status: 'accepted',
    };
  }

  async markUnknownTimeout(data: TimeoutUnknownDto) {
    const olderThanHours = data.olderThanHours ?? 72;
    const cutoff = new Date(Date.now() - olderThanHours * 60 * 60 * 1000);
    const candidates = await this.prisma.smsMessageRecord.findMany({
      where: {
        status: 'unknown',
        deliveredAt: { lte: cutoff },
      },
      select: { id: true, batchTaskId: true },
      take: 10000,
    });
    await this.prisma.smsMessageRecord.updateMany({
      where: { id: { in: candidates.map((candidate) => candidate.id) } },
      data: { status: 'timeout', timeoutAt: new Date(), errorMessage: '72小时未收到明确回执，自动转超时' },
    });
    for (const candidate of candidates) {
      const message = await this.prisma.smsMessageRecord.findUnique({ where: { id: candidate.id } });
      if (message?.tenantId) {
        await this.refundMessage(message as typeof message & { tenantId: string }, '72小时未收到明确回执，自动超时退款');
      }
    }
    for (const batchTaskId of new Set(candidates.map((candidate) => candidate.batchTaskId).filter((value): value is string => Boolean(value)))) {
      await this.refreshTaskProgress(batchTaskId);
    }
    return { timeout: candidates.length };
  }

  private async submitMessageToGateway(
    message: {
      id: string;
      tenantId: string;
      batchTaskId: string;
      applicationId?: string | null;
      templateId?: string | null;
      messageId: string;
      phoneNumber: string;
      content: string;
      billingUnits: number;
      queuePriority?: string | null;
      template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
      signature?: { id?: string | null; name?: string | null } | null;
    },
    routed: RoutedChannel,
    attempt: number,
  ) {
    const channel = routed.channel;
    await this.ensureSignatureReportedForChannel(message, channel.id);
    await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
    const submitId = `SUB-${randomUUID()}`;
    const session = await this.prisma.cmppSubmitSession.upsert({
      where: { sessionNo: `OPEN-${channel.id}` },
      update: { submitTotal: { increment: 1 } },
      create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
    });
    await this.prisma.smsSubmitRecord.create({
      data: {
        tenantId: message.tenantId,
        batchTaskId: message.batchTaskId,
        messageRecordId: message.id,
        channelId: channel.id,
        sessionId: session.id,
        submitId,
        submitStatus: 'queued',
      },
    });
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: {
        channelId: channel.id,
        carrier: routed.carrier,
        province: routed.province,
        submitId,
        status: 'submit_queued',
        submitStatus: 'queued',
        receiptStatus: null,
        errorCode: null,
        errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined,
      },
    });
    const command = {
      schemaVersion: 'v1',
      messageType: 'SubmitCommand',
      traceId: randomUUID(),
      messageId: message.messageId,
      channelId: channel.id,
      createdAt: new Date().toISOString(),
      tenantId: message.tenantId,
      applicationId: message.applicationId ?? 'unknown',
      taskId: message.batchTaskId,
      submitId,
      queuePriority: normalizeQueuePriority(message.queuePriority),
      phoneNumber: message.phoneNumber,
      content: message.content,
      signature: message.template?.signature?.name ?? message.signature?.name ?? 'SMS',
      templateId: message.templateId ?? 'unknown',
      billingUnits: message.billingUnits,
      route: {
        channelCode: channel.code,
        cmppAccountCode: channel.account,
        priority: attempt,
        rateLimitPerSecond: channel.rateLimitPerSecond,
        carrier: routed.carrier,
        province: routed.province ?? undefined,
        scope: routed.routeScope,
        groupId: routed.groupId,
      },
      cmpp: {
        serviceId: channel.config && typeof channel.config === 'object' && 'serviceId' in channel.config
          ? String(channel.config.serviceId)
          : 'SMS',
        srcId: channel.srcId,
        extensionDigits: getNonNegativeConfigInteger(channel.config, 'extensionDigits', 0),
        registeredDelivery: 1,
        msgFmt: 8,
      },
      upstream: {
        gatewayHost: channel.gatewayHost,
        gatewayPort: channel.gatewayPort,
        account: channel.account,
        passwordCipher: channel.passwordCipher,
        cmppVersion: channel.cmppVersion,
        desiredConnections: getPositiveConfigInteger(channel.config, 'desiredConnections', 1),
        windowSize: getPositiveConfigInteger(channel.config, 'windowSize', 16),
      },
      retry: { attempt, maxAttempts: 1 },
    };
    await this.getGatewayQueue().add('submit-command', command);
    await this.publishGatewaySubmitCommand(command);
    await this.refreshTaskProgress(message.batchTaskId);
    return { submitted: true, messageRecordId: message.id, channelId: channel.id, attempt };
  }

  private async retryMessageIfAllowed(
    message: {
      id: string;
      tenantId: string;
      batchTaskId: string;
      applicationId?: string | null;
      templateId?: string | null;
      messageId: string;
      phoneNumber: string;
      content: string;
      billingUnits: number;
      queuedAt?: Date;
    },
    reason: string,
  ) {
    const attempts = await this.prisma.smsSubmitRecord.findMany({
      where: { messageRecordId: message.id },
      orderBy: { createdAt: 'asc' },
      take: 200,
    });
    const attemptedChannelIds = attempts.map((attempt) => attempt.channelId);
    const ageMinutes = (Date.now() - new Date(message.queuedAt ?? Date.now()).getTime()) / 60_000;
    if (ageMinutes >= 72 * 60) {
      return null;
    }
    const route = await this.findApplicationRoute(message.tenantId, message.applicationId ?? undefined, await this.identifyCarrier(message.phoneNumber));
    const retryTimeLimitMinutes = Math.min(route.group.retryTimeLimitMinutes ?? route.group.retryTimeLimitHours * 60, 72 * 60);
    if (!route.group.retryEnabled || ageMinutes >= retryTimeLimitMinutes) {
      return null;
    }
    try {
      const routed = await this.selectChannelForMessage(message, {
        forceNational: true,
        excludeChannelIds: attemptedChannelIds,
      });
      await this.prisma.smsMessageRecord.update({
        where: { id: message.id },
        data: { errorMessage: reason },
      });
      return await this.submitMessageToGateway(message, routed, attempts.length);
    } catch {
      return null;
    }
  }

  private async selectChannelForMessage(
    message: { id: string; tenantId: string; applicationId?: string | null; templateId?: string | null; phoneNumber: string; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null },
    options: { forceNational?: boolean; excludeChannelIds?: string[] } = {},
  ): Promise<RoutedChannel> {
    if (!message.applicationId) {
      throw new BadRequestException('短信应用未配置，无法选择通道组');
    }
    const carrier = await this.identifyCarrier(message.phoneNumber);
    const province = await this.identifyProvince(message.phoneNumber);
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: { carrier, province },
    });
    const route = await this.findApplicationRoute(message.tenantId, message.applicationId, carrier);
    const excluded = new Set(options.excludeChannelIds ?? []);
    const signatureId = await this.resolveMessageSignatureId(message);
    if (!signatureId) throw new BadRequestException('短信签名未配置，无法选择已报备通道');
    const approvedTasks = await this.prisma.channelSignatureReportTask.findMany({
      where: { signatureId, reportType: 'signature', status: 'approved', channelId: { in: route.group.items.map((item) => item.channelId) } },
      select: { channelId: true },
    });
    const approvedChannelIds = new Set(approvedTasks.map((task) => task.channelId));
    const items = route.group.items.filter((item) =>
      !excluded.has(item.channelId)
      && approvedChannelIds.has(item.channelId)
      && normalizeCarrier(item.carrier) === carrier
      && isCarrierCompatible(item.channel.carrier, carrier),
    );
    const provinceCandidates = options.forceNational ? [] : items.filter((item) => isProvinceChannel(item, province));
    const nationalCandidates = items.filter((item) => isNationalChannel(item));
    const selected = [...provinceCandidates, ...nationalCandidates].find((item) => this.isChannelSendAvailable(item.channel));
    if (!selected) {
      throw new NotFoundException('无已报备通过且在线的可用通道');
    }
    return {
      channel: selected.channel,
      carrier,
      province,
      groupId: route.groupId,
      routeScope: isNationalChannel(selected) ? 'national' : 'province',
    };
  }

  private async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string) {
    const route = await this.prisma.channelRouteRule.findFirst({
      where: {
        status: 'active',
        tenantId,
        applicationId,
        carrier,
        channelId: null,
        province: null,
      },
      include: { group: { include: { items: { include: { channel: { include: { connectionStates: true } } }, orderBy: { priority: 'asc' } } } } },
      orderBy: { priority: 'asc' },
    });
    if (!route) {
      throw new NotFoundException('企业应用未配置对应运营商通道组');
    }
    if (route.group.status !== 'active') {
      throw new BadRequestException('企业应用绑定的通道组已停用');
    }
    if (normalizeCarrier(route.group.carrier) !== carrier) {
      throw new BadRequestException('企业应用绑定的通道组运营商与路由规则不一致');
    }
    return route;
  }

  private async identifyCarrier(phoneNumber: string) {
    const rules = await this.prisma.phoneCarrierRule.findMany({
      where: { status: 'active' },
      orderBy: [{ priority: 'asc' }, { createdAt: 'asc' }],
    });
    for (const rule of rules) {
      try {
        if (new RegExp(rule.pattern).test(phoneNumber)) {
          return normalizeCarrier(rule.carrier);
        }
      } catch {
        continue;
      }
    }
    return 'mobile';
  }

  private async identifyProvince(phoneNumber: string) {
    for (let length = Math.min(7, phoneNumber.length); length >= 3; length -= 1) {
      const segment = await this.prisma.phoneSegment.findUnique({ where: { prefix: phoneNumber.slice(0, length) } });
      if (segment?.province) {
        return segment.province;
      }
    }
    return null;
  }

  private isChannelSendAvailable(channel: { status: string; connectionStates?: Array<{ status: string; currentConnections: number; desiredConnections: number }> }) {
    if (channel.status !== 'active') {
      return false;
    }
    return (channel.connectionStates ?? []).some((connection) =>
      connection.desiredConnections > 0 && connection.currentConnections > 0 && connection.status === 'connected',
    );
  }

  private async resolveUnitPrice(tenantId: string, applicationId?: string) {
    if (!applicationId) {
      return 0;
    }
    const application = await this.prisma.smsApplication.findUnique({
      where: { id: applicationId },
      select: { tenantId: true, customerUnitPrice: true },
    });
    if (!application || application.tenantId !== tenantId) {
      return 0;
    }
    return application.customerUnitPrice ?? 0;
  }

  private async resolveQueuePriority(tenantId: string, applicationId?: string): Promise<QueuePriority> {
    if (!applicationId) {
      return 'normal';
    }
    const application = await this.prisma.smsApplication.findUnique({
      where: { id: applicationId },
      select: { tenantId: true, queuePriority: true },
    });
    if (!application || application.tenantId !== tenantId) {
      return 'normal';
    }
    return normalizeQueuePriority(application.queuePriority);
  }

  private findInboundApplication(account: string) {
    return this.prisma.smsApplication.findFirst({
      where: { cmppAccount: account },
      include: {
        tenant: true,
        ipAllowlist: true,
      },
    });
  }

  private resolveInboundTemplateCandidate(applicationId: string, content: string) {
    return this.prisma.smsTemplate.findFirst({
      where: {
        applicationId,
        content,
      },
      include: { signature: true },
      orderBy: { updatedAt: 'desc' },
    });
  }

  private resolveInboundSignatureCandidate(applicationId: string, content: string) {
    const match = content.match(/^【[^】]+】/);
    if (!match?.[0]) return null;
    return this.prisma.smsSignature.findFirst({
      where: {
        applicationId,
        name: match[0],
        auditStatus: 'approved',
      },
      orderBy: { updatedAt: 'desc' },
    });
  }

  private async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, signatureId: string) {
    await this.prisma.smsMessageRecord.update({
      where: { id: messageRecordId },
      data: { reviewTaskId, signatureId, status: 'pending_review' },
    });
    return this.prisma.smsSendTask.findUnique({ where: { id: reviewTaskId } });
  }

  private async recordCmppFailureReceipt(
    message: {
      id: string;
      tenantId?: string | null;
      batchTaskId?: string | null;
      applicationId?: string | null;
      messageId: string;
      phoneNumber: string;
      cmppSubmitSequenceId?: string | 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}` },
    });
    if (existing) return existing;
    const deliveredAt = new Date();
    await this.prisma.smsMessageRecord.update({
      where: { id: message.id },
      data: { status: 'failed', receiptStatus: 'undelivered', errorCode, errorMessage: reason, deliveredAt },
    });
    const receipt = await this.prisma.smsReceiptRecord.create({
      data: {
        tenantId: message.tenantId,
        batchTaskId: message.batchTaskId,
        messageRecordId: message.id,
        messageId: message.messageId,
        gatewayMessageId: `PLATFORM:${message.messageId}`,
        receiptStatus: 'undelivered',
        rawStatus: 'REJECTD',
        errorCode,
        deliveredAt,
      },
    });
    await this.queueAndTryDownstreamDelivery({
      tenantId: message.tenantId,
      applicationId: message.applicationId,
      messageRecordId: message.id,
      messageId: message.messageId,
      deliveryType: 'receipt',
      payload: {
        messageId: message.messageId,
        gatewayMessageId: `PLATFORM:${message.messageId}`,
        phoneNumber: message.phoneNumber,
        receiptStatus: 'undelivered',
        rawStatus: 'REJECTD',
        errorCode,
        errorMessage: reason,
        submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined,
        deliveredAt: deliveredAt.toISOString(),
      },
    });
    if (message.batchTaskId) await this.refreshTaskProgress(message.batchTaskId);
    return receipt;
  }

  private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) {
    const tenant = await this.prisma.tenant.findUnique({ where: { id: tenantId } });
    if (!tenant || tenant.status !== 'active') {
      throw new BadRequestException('企业客户不存在或已停用');
    }
    if (tenant.certificationStatus !== 'approved') {
      throw new BadRequestException('企业认证未通过，不能发送短信');
    }
    if (!applicationId) {
      return;
    }
    const application = await this.prisma.smsApplication.findUnique({ where: { id: applicationId } });
    if (!application || application.tenantId !== tenantId || application.status !== 'active') {
      throw new BadRequestException('短信应用不存在或已停用');
    }
    if (!application.interfaceEnabled) {
      throw new BadRequestException('短信应用接口未开通，不能发送短信');
    }
    if (!templateId) {
      return;
    }
    const template = await this.prisma.smsTemplate.findUnique({
      where: { id: templateId },
      include: { signature: true },
    });
    if (!template || template.tenantId !== tenantId || template.applicationId !== applicationId || template.auditStatus !== 'approved') {
      throw new BadRequestException('短信模板不存在、未通过审核或不属于当前应用');
    }
    if (!template.signature || template.signature.auditStatus !== 'approved') {
      throw new BadRequestException('短信签名未审核通过');
    }
  }

  private async chargeAcceptedMessage(message: {
    tenantId: string;
    applicationId?: string | null;
    batchTaskId: string;
    messageId: string;
    phoneNumber: string;
    content: string;
    billingUnits: number;
    unitPrice: number;
    amountCents: number;
  }) {
    const amountCents = message.amountCents ?? 0;
    const billingUnits = message.billingUnits ?? 0;
    const exists = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId } });
    if (exists?.billingStatus === 'charged') {
      return;
    }
    if (amountCents > 0) {
      await this.billing.release({
        tenantId: message.tenantId,
        amountCents,
        relatedType: 'sms_batch_task',
        relatedId: message.batchTaskId,
        remark: `短信 ${message.messageId} 提交成功释放冻结并转扣费`,
      });
    }
    const transaction = await this.billing.charge({
      tenantId: message.tenantId,
      amountCents,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark: '提交成功扣费',
    });
    const data = {
      tenantId: message.tenantId,
      applicationId: message.applicationId ?? undefined,
      taskId: message.batchTaskId,
      messageId: message.messageId,
      phoneNumber: message.phoneNumber,
      contentLength: [...message.content].length,
      billingUnits,
      unitPrice: message.unitPrice ?? 0,
      amountCents,
      billingStatus: 'charged',
      transactionId: transaction.id,
    };
    if (exists) {
      await this.prisma.smsBillingRecord.update({ where: { id: exists.id }, data });
      return;
    }
    await this.prisma.smsBillingRecord.create({ data });
  }

  private async releaseMessageReservation(
    message: { tenantId: string; batchTaskId: string; messageId: string; amountCents: number; billingUnits: number },
    remark: string,
  ) {
    if ((message.amountCents ?? 0) <= 0) {
      return;
    }
    const charged = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId, billingStatus: 'charged' } });
    if (charged) {
      return;
    }
    const released = await this.prisma.accountTransaction.findFirst({
      where: { relatedType: 'sms_message_record', relatedId: message.messageId, transactionType: 'released' },
    });
    if (released) {
      return;
    }
    await this.billing.release({
      tenantId: message.tenantId,
      amountCents: message.amountCents,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark: `${remark}: ${message.messageId}`,
    });
  }

  private async refundMessage(
    message: { tenantId: string; messageId: string; amountCents: number; billingUnits: number },
    remark: string,
  ) {
    if ((message.amountCents ?? 0) <= 0) {
      return;
    }
    const refunded = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId, billingStatus: 'refunded' } });
    if (refunded) {
      return;
    }
    const charged = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId, billingStatus: 'charged' } });
    if (!charged) {
      return;
    }
    const transaction = await this.billing.refund({
      tenantId: message.tenantId,
      amountCents: message.amountCents,
      relatedType: 'sms_message_record',
      relatedId: message.messageId,
      remark,
    });
    await this.prisma.smsBillingRecord.updateMany({
      where: { messageId: message.messageId },
      data: { billingStatus: 'refunded', transactionId: transaction.id },
    });
  }

  private async ensureSignatureReportedForChannel(
    message: {
      id: string;
      templateId?: string | null;
      template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
      signature?: { id?: string | null; name?: string | null } | null;
    },
    channelId: string,
  ) {
    const signatureId = await this.resolveMessageSignatureId(message);
    if (!signatureId) {
      throw new BadRequestException('短信签名未配置，不能提交到通道');
    }
    const reportTask = await this.prisma.channelSignatureReportTask.findFirst({
      where: { signatureId, channelId, reportType: 'signature', status: 'approved' },
      select: { id: true },
    });
    if (!reportTask) {
      throw new BadRequestException('短信签名未在最终通道报备通过');
    }
  }

  private async resolveMessageSignatureId(message: { templateId?: string | null; template?: { signature?: { id?: string | null } | null } | null; signature?: { id?: string | null } | null }) {
    const direct = message.template?.signature?.id ?? message.signature?.id ?? null;
    if (direct || !message.templateId) return direct;
    const template = await this.prisma.smsTemplate.findUnique({ where: { id: message.templateId }, include: { signature: true } });
    return template?.signature?.id ?? null;
  }

  private async waitForChannelRateLimit(channelId: string, tps: number) {
    const redis = this.getRedis();
    for (;;) {
      const bucket = `rate:channel:${channelId}:${Math.floor(Date.now() / 1000)}`;
      const count = await redis.incr(bucket);
      if (count === 1) {
        await redis.expire(bucket, 2);
      }
      if (count <= Math.max(1, tps)) {
        return;
      }
      await sleep(100);
    }
  }

  private async refreshTaskProgress(batchTaskId: string) {
    const groups = await this.prisma.smsMessageRecord.groupBy({
      by: ['status'],
      where: { batchTaskId },
      _count: { _all: true },
    });
    const count = (statuses: string[]) =>
      groups.filter((group) => statuses.includes(group.status)).reduce((sum, group) => sum + group._count._all, 0);
    const progressTotal = groups.reduce((sum, group) => sum + group._count._all, 0);
    const submittedTotal = count(['submit_queued', 'submitted', 'delivered', 'failed', 'unknown', 'timeout']);
    const successTotal = count(['delivered']);
    const failedTotal = count(['submit_failed', 'failed']);
    const unknownTotal = count(['unknown']);
    const timeoutTotal = count(['timeout']);
    const doneTotal = successTotal + failedTotal + timeoutTotal;
    const status = progressTotal > 0 && doneTotal >= progressTotal ? 'finished' : submittedTotal > 0 ? 'sending' : 'queued';
    await this.prisma.smsBatchTask.update({
      where: { id: batchTaskId },
      data: { progressTotal, submittedTotal, successTotal, failedTotal, unknownTotal, timeoutTotal, status },
    });
  }

  private smsMessageSegmentAuditDelegate() {
    return (this.prisma as PrismaService & {
      smsMessageSegmentAudit: {
        upsert: (args: Record<string, unknown>) => Promise<any>;
        updateMany: (args: Record<string, unknown>) => Promise<{ count: number }>;
        findFirst: (args: Record<string, unknown>) => Promise<any | null>;
      };
    }).smsMessageSegmentAudit;
  }

  private async recordSubmitSegments(
    message: {
      id: string;
      tenantId?: string | null;
      batchTaskId?: string | null;
      channelId?: string | null;
      submitId?: string | null;
      billingUnits?: number | null;
    },
    data: GatewaySubmitResultDto,
    submittedAt: Date,
  ) {
    const segmentAudits = this.smsMessageSegmentAuditDelegate();
    const submitRecord = await this.prisma.smsSubmitRecord.findFirst({
      where: {
        messageRecordId: message.id,
        OR: [
          data.submitId ? { submitId: data.submitId } : undefined,
          data.gatewayMessageId ? { gatewayMessageId: data.gatewayMessageId } : undefined,
        ].filter(Boolean) as Array<{ submitId?: string; gatewayMessageId?: string }>,
      },
      orderBy: { createdAt: 'desc' },
    });
    const submitId = data.submitId ?? submitRecord?.submitId ?? message.submitId ?? `SUB-AUDIT-${message.id}`;
    const attempt = submitRecord
      ? Math.max(0, await this.prisma.smsSubmitRecord.count({
        where: {
          messageRecordId: message.id,
          createdAt: { lte: submitRecord.createdAt },
        },
      }) - 1)
      : 0;
    const fallbackSegments = [{
      segmentTotal: Math.max(1, Number(message.billingUnits ?? 1)),
      segmentIndex: 1,
      sequenceId: data.sequenceId,
      gatewayMessageId: data.gatewayMessageId,
      submitStatus: data.submitStatus,
      errorCode: data.errorCode,
      errorMessage: data.errorMessage,
      submittedAt: data.submittedAt,
    }];
    const segments = data.segments && data.segments.length > 0 ? data.segments : fallbackSegments;
    const segmentTotal = Math.max(1, ...segments.map((item) => Number(item.segmentTotal ?? segments.length ?? 1)));
    await Promise.all(segments.map((segment, index) => {
      const segmentIndex = Math.max(1, Number(segment.segmentIndex ?? index + 1));
      const status = segment.submitStatus ?? data.submitStatus;
      return segmentAudits.upsert({
        where: {
          messageRecordId_submitId_segmentIndex: {
            messageRecordId: message.id,
            submitId,
            segmentIndex,
          },
        },
        update: {
          submitRecordId: submitRecord?.id ?? null,
          channelId: data.channelId ?? message.channelId ?? null,
          attempt,
          segmentTotal,
          sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
          gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
          submitStatus: status,
          errorCode: segment.errorCode ?? data.errorCode ?? null,
          errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
          submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
        },
        create: {
          tenantId: message.tenantId,
          batchTaskId: message.batchTaskId,
          messageRecordId: message.id,
          submitRecordId: submitRecord?.id ?? null,
          channelId: data.channelId ?? message.channelId ?? null,
          submitId,
          attempt,
          segmentTotal,
          segmentIndex,
          sequenceId: segment.sequenceId ?? data.sequenceId ?? null,
          gatewayMessageId: segment.gatewayMessageId ?? data.gatewayMessageId ?? null,
          submitStatus: status,
          compensationType: submitRecord && submitRecord.submitId !== message.submitId ? 'retry_submit' : null,
          errorCode: segment.errorCode ?? data.errorCode ?? null,
          errorMessage: segment.errorMessage ?? data.errorMessage ?? null,
          submittedAt: segment.submittedAt ? new Date(segment.submittedAt) : submittedAt,
        },
      });
    }));
  }

  private async recordReceiptSegment(
    message: {
      id: string;
      tenantId?: string | null;
      batchTaskId?: string | null;
      channelId?: string | null;
      submitId?: string | null;
      billingUnits?: number | null;
    },
    data: GatewayReceiptEventDto,
    deliveredAt: Date,
    submitRecordId?: string,
  ) {
    const segmentAudits = this.smsMessageSegmentAuditDelegate();
    const updated = await segmentAudits.updateMany({
      where: {
        messageRecordId: message.id,
        gatewayMessageId: data.gatewayMessageId,
      },
      data: {
        receiptStatus: data.receiptStatus,
        rawStatus: data.rawStatus,
        errorCode: data.errorCode ?? null,
        deliveredAt,
      },
    });
    if (updated.count > 0) {
      return;
    }
    const submitRecord = submitRecordId
      ? await this.prisma.smsSubmitRecord.findUnique({ where: { id: submitRecordId } })
      : await this.prisma.smsSubmitRecord.findFirst({
        where: { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId },
        orderBy: { createdAt: 'desc' },
      });
    await segmentAudits.upsert({
      where: {
        messageRecordId_submitId_segmentIndex: {
          messageRecordId: message.id,
          submitId: submitRecord?.submitId ?? message.submitId ?? `RECEIPT-AUDIT-${data.gatewayMessageId}`,
          segmentIndex: 1,
        },
      },
      update: {
        submitRecordId: submitRecord?.id ?? submitRecordId ?? null,
        channelId: data.channelId ?? message.channelId ?? null,
        sequenceId: data.sequenceId ?? null,
        gatewayMessageId: data.gatewayMessageId,
        receiptStatus: data.receiptStatus,
        rawStatus: data.rawStatus,
        errorCode: data.errorCode ?? null,
        deliveredAt,
      },
      create: {
        tenantId: message.tenantId,
        batchTaskId: message.batchTaskId,
        messageRecordId: message.id,
        submitRecordId: submitRecord?.id ?? submitRecordId ?? null,
        channelId: data.channelId ?? message.channelId ?? null,
        submitId: submitRecord?.submitId ?? message.submitId ?? `RECEIPT-AUDIT-${data.gatewayMessageId}`,
        attempt: 0,
        segmentTotal: Math.max(1, Number(message.billingUnits ?? 1)),
        segmentIndex: 1,
        sequenceId: data.sequenceId ?? null,
        gatewayMessageId: data.gatewayMessageId,
        submitStatus: submitRecord?.submitStatus ?? 'accepted',
        receiptStatus: data.receiptStatus,
        rawStatus: data.rawStatus,
        compensationType: 'receipt_recovered',
        errorCode: data.errorCode ?? null,
        deliveredAt,
      },
    });
  }

  private async findMessageByGatewayEvent(messageId?: string, gatewayMessageId?: string) {
    const conditions = [{ messageId }, gatewayMessageId ? { gatewayMessageId } : undefined].filter(
      Boolean,
    ) as Array<{
      messageId?: string;
      gatewayMessageId?: string;
    }>;
    if (conditions.length === 0) {
      return null;
    }
    return this.prisma.smsMessageRecord.findFirst({
      where: {
        OR: conditions,
      },
    });
  }

  private async requireMessageByGatewayEvent(messageId?: string, gatewayMessageId?: string) {
    const message = await this.findMessageByGatewayEvent(messageId, gatewayMessageId);
    if (!message) {
      throw new NotFoundException('SMS message record not found');
    }
    return message;
  }

  private async resolveReceiptMessage(data: GatewayReceiptEventDto) {
    const exactMessage = await this.findMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
    if (exactMessage) {
      return {
        message: exactMessage,
        messageId: exactMessage.messageId,
      };
    }

    const phoneNumber = data.phoneNumber?.trim();
    if (!phoneNumber) {
      throw new NotFoundException('SMS message record not found');
    }

    const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
    const submittedAfter = new Date(deliveredAt.getTime() - 72 * 60 * 60 * 1000);
    const candidates = await this.prisma.smsSubmitRecord.findMany({
      where: {
        channelId: data.channelId,
        gatewayMessageId: null,
        submitStatus: 'timeout',
        submittedAt: {
          gte: submittedAfter,
          lte: deliveredAt,
        },
        messageRecord: {
          phoneNumber,
        },
      },
      include: {
        messageRecord: true,
      },
      orderBy: {
        submittedAt: 'desc',
      },
      take: 10,
    });

    if (candidates.length !== 1 || !candidates[0]?.messageRecord) {
      throw new NotFoundException('SMS message record not found');
    }

    return {
      message: candidates[0].messageRecord,
      messageId: candidates[0].messageRecord.messageId,
      submitRecordId: candidates[0].id,
    };
  }

  private getSendQueue(): Queue<SendJob, unknown, 'send-message'> {
    if (!this.sendQueue) {
      this.sendQueue = new Queue<SendJob, unknown, 'send-message'>(SEND_QUEUE, { connection: bullmqConnection() });
    }
    return this.sendQueue;
  }

  private getGatewayQueue(): Queue {
    if (!this.gatewayQueue) {
      this.gatewayQueue = new Queue(GATEWAY_SUBMIT_QUEUE, { connection: bullmqConnection() });
    }
    return this.gatewayQueue;
  }

  private getRedis() {
    if (!this.redis) {
      this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', {
        maxRetriesPerRequest: null,
      });
    }
    return this.redis;
  }

  private 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(() => ({}));
  }

  private async publishGatewaySubmitCommand(command: unknown) {
    return this.getRedis().xadd(
      process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM,
      '*',
      'messageType',
      'SubmitCommand',
      'data',
      JSON.stringify(command),
    );
  }

}

function statusFromRisk(status: string, scheduled: boolean) {
  if (status === 'rejected') {
    return 'rejected';
  }
  if (status === 'pending_review') {
    return 'pending_review';
  }
  if (scheduled) {
    return 'scheduled';
  }
  return 'ready';
}

function parseSchedule(data: CreateBatchTaskDto) {
  if (data.sendMode !== 'scheduled' && !data.scheduledAt) {
    return { scheduledAt: null };
  }
  if (!data.scheduledAt) {
    throw new BadRequestException('定时发送必须提供 scheduledAt');
  }
  const scheduledAt = new Date(data.scheduledAt);
  if (Number.isNaN(scheduledAt.getTime())) {
    throw new BadRequestException('scheduledAt 时间格式无效');
  }
  if (scheduledAt.getTime() <= Date.now()) {
    throw new BadRequestException('scheduledAt 必须晚于当前时间');
  }
  return { scheduledAt };
}

function isObjectRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === 'object' && value !== null && !Array.isArray(value);
}

function asDateOrNull(value?: string | null) {
  if (!value) {
    return null;
  }
  const parsed = new Date(value);
  return Number.isNaN(parsed.getTime()) ? null : parsed;
}

function downstreamRetryDelayMs(retryCount = 1) {
  const base = downstreamRetryBaseDelayMs();
  const max = downstreamRetryMaxDelayMs();
  const attempt = Math.max(1, Math.floor(retryCount));
  const delay = base * Math.pow(2, Math.max(0, attempt - 1));
  return Math.min(delay, max);
}

function downstreamAckTimeoutMs() {
  const configured = Number(process.env.CMPP_DOWNSTREAM_ACK_TIMEOUT_SECONDS ?? 30);
  return Math.max(5, Number.isFinite(configured) ? configured : 30) * 1000;
}

function downstreamRetryBaseDelayMs() {
  const value = Number(process.env.CMPP_DOWNSTREAM_RETRY_DELAY_MS ?? DEFAULT_DOWNSTREAM_RETRY_DELAY_MS);
  return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_RETRY_DELAY_MS;
}

function downstreamRetryMaxDelayMs() {
  const value = Number(process.env.CMPP_DOWNSTREAM_RETRY_MAX_DELAY_MS ?? DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS);
  return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS;
}

function downstreamMaxRetries() {
  const value = Number(process.env.CMPP_DOWNSTREAM_MAX_RETRIES ?? DEFAULT_DOWNSTREAM_MAX_RETRIES);
  return Number.isFinite(value) && value > 0 ? Math.floor(value) : DEFAULT_DOWNSTREAM_MAX_RETRIES;
}

function parseImportRows(content: string, delimiter?: ',' | '\t') {
  const normalized = content.replace(/^\uFEFF/, '');
  const lines = normalized.split(/\r?\n/).filter((line) => line.trim().length > 0);
  if (lines.length === 0) {
    return [];
  }
  const firstDelimiter = delimiter ?? (lines[0].includes(',') ? ',' : '\t');
  const firstCells = splitImportLine(lines[0], firstDelimiter);
  const hasHeader = firstCells.some((cell) => ['phone', 'phoneNumber', 'mobile', '手机号'].includes(cell));
  const headers = hasHeader ? firstCells : ['phoneNumber'];
  const dataLines = hasHeader ? lines.slice(1) : lines;
  return dataLines.map((line, index) => {
    const cells = splitImportLine(line, firstDelimiter);
    const row: { rowNumber: number; phoneNumber?: string; variables: Record<string, string> } = {
      rowNumber: (hasHeader ? index + 2 : index + 1),
      phoneNumber: hasHeader ? cellByHeader(headers, cells, ['phone', 'phoneNumber', 'mobile', '手机号']) : cells[0],
      variables: {},
    };
    headers.forEach((header, cellIndex) => {
      if (!['phone', 'phoneNumber', 'mobile', '手机号'].includes(header)) {
        row.variables[header] = cells[cellIndex] ?? '';
      }
    });
    return row;
  });
}

function splitImportLine(line: string, delimiter: ',' | '\t') {
  return line.split(delimiter).map((cell) => cell.trim().replace(/^"|"$/g, ''));
}

function cellByHeader(headers: string[], cells: string[], candidates: string[]) {
  const index = headers.findIndex((header) => candidates.includes(header));
  return index >= 0 ? cells[index] : undefined;
}

function normalizeCarrier(carrier?: string | null) {
  const value = String(carrier ?? '').trim().toLowerCase();
  if (['mobile', 'cmcc', '移动', '中国移动'].includes(value)) return 'mobile';
  if (['unicom', 'cucc', '联通', '中国联通'].includes(value)) return 'unicom';
  if (['telecom', 'ctcc', '电信', '中国电信'].includes(value)) return 'telecom';
  if (['all', 'tri', '三网', '全网'].includes(value)) return 'all';
  return value || 'mobile';
}

function normalizeQueuePriority(queuePriority?: string | null): QueuePriority {
  return queuePriority === 'priority' ? 'priority' : 'normal';
}

function getPositiveConfigInteger(config: unknown, key: string, fallback: number) {
  if (config && typeof config === 'object' && !Array.isArray(config) && key in config) {
    const value = Number((config as Record<string, unknown>)[key]);
    if (Number.isInteger(value) && value > 0) {
      return value;
    }
  }
  return fallback;
}

function getNonNegativeConfigInteger(config: unknown, key: string, fallback: number) {
  if (config && typeof config === 'object' && !Array.isArray(config) && key in config) {
    const value = Number((config as Record<string, unknown>)[key]);
    if (Number.isInteger(value) && value >= 0) {
      return value;
    }
  }
  return fallback;
}

function isCarrierCompatible(channelCarrier: string | null | undefined, targetCarrier: string) {
  const normalized = normalizeCarrier(channelCarrier);
  return normalized === 'all' || normalized === targetCarrier;
}

function normalizeRegion(region?: string | null) {
  return String(region ?? '').replace(/省|市|自治区|壮族|回族|维吾尔/g, '').trim();
}

function isNationalChannel(item: { province?: string | null; channel: { sendRegion?: string | null } }) {
  const itemProvince = normalizeRegion(item.province);
  const sendRegion = normalizeRegion(item.channel.sendRegion);
  return !itemProvince || itemProvince === '全国' || !sendRegion || sendRegion === '全国';
}

function isProvinceChannel(item: { province?: string | null; channel: { sendRegion?: string | null } }, province?: string | null) {
  if (!province) {
    return false;
  }
  const target = normalizeRegion(province);
  const itemProvince = normalizeRegion(item.province);
  const sendRegion = normalizeRegion(item.channel.sendRegion);
  return itemProvince === target || sendRegion === target;
}

function bullmqConnection() {
  const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
  return {
    host: redisUrl.hostname,
    port: Number(redisUrl.port || 6379),
    username: redisUrl.username || undefined,
    password: redisUrl.password || undefined,
    maxRetriesPerRequest: null,
  };
}

function matchesApplicationSecret(data: GatewayInboundAuthDto, secretHash: string) {
  if (data.authSource && data.timestamp !== undefined) {
    const expected = createHash('md5')
      .update(Buffer.concat([
        Buffer.from(octetString(data.account, 6), 'binary'),
        Buffer.alloc(9),
        Buffer.from(secretHash),
        Buffer.from(String(data.timestamp).padStart(10, '0')),
      ]))
      .digest('base64');
    return expected === data.authSource;
  }
  if (!data.password) {
    return false;
  }
  return data.password === secretHash || createHash('sha256').update(data.password).digest('hex') === secretHash;
}

function octetString(value: string, fixedLength: number) {
  if (value.length === fixedLength) {
    return value;
  }
  if (value.length > fixedLength) {
    return value.slice(value.length - fixedLength);
  }
  return value + '\0'.repeat(fixedLength - value.length);
}

function hasRecoveryAuditStateChanged(
  previous: Record<string, unknown> | null,
  current: Record<string, unknown>,
) {
  if (!previous) {
    return true;
  }
  return ['state', 'gatewayInstanceId', 'lockOwner', 'failureCategory', 'lastError', 'lastSkipReason']
    .some((key) => (previous[key] ?? null) !== (current[key] ?? null));
}

function normalizeRecoveryFailureCategory(data: GatewayDownstreamRecoveryStatusDto) {
  const explicit = String(data.failureCategory ?? '').trim();
  if (explicit) {
    return explicit;
  }
  if (data.state === 'success' || data.state === 'running') {
    return null;
  }
  if (data.lastSkipReason === 'backoff') {
    return 'backoff';
  }
  if (data.lastSkipReason === 'locked') {
    return 'lock_contended';
  }
  if (data.lastSkipReason === 'lock_lost') {
    return 'lock_lost';
  }
  if (data.state === 'waiting_connection') {
    return 'client_disconnected';
  }
  if (data.state === 'partial') {
    return 'partial_delivery_failed';
  }
  if (data.state === 'failed' && data.lastError) {
    return 'flush_failed';
  }
  return data.state ? 'unknown' : null;
}

function isApplicationIpAllowed(remoteIp: string, allowlist: string[]) {
  const normalizedRemoteIp = normalizeIp(remoteIp);
  if (allowlist.length === 0) {
    return true;
  }
  return allowlist.some((rule) => ipMatchesRule(normalizedRemoteIp, rule));
}

function ipMatchesRule(remoteIp: string, rule: string) {
  const normalizedRule = normalizeIp(rule.trim());
  if (!normalizedRule) {
    return false;
  }
  if (!normalizedRule.includes('/')) {
    return remoteIp === normalizedRule;
  }
  const [network, prefixText] = normalizedRule.split('/');
  const prefix = Number(prefixText);
  if (!Number.isInteger(prefix) || prefix < 0 || prefix > 32 || isIP(remoteIp) !== 4 || isIP(network) !== 4) {
    return false;
  }
  const mask = prefix === 0 ? 0 : (0xffffffff << (32 - prefix)) >>> 0;
  return (ipv4ToInt(remoteIp) & mask) === (ipv4ToInt(network) & mask);
}

function normalizeIp(value: string) {
  return value.replace(/^::ffff:/, '').trim();
}

function ipv4ToInt(value: string) {
  return value.split('.').reduce((result, part) => ((result << 8) + Number(part)) >>> 0, 0);
}