import { BadRequestException, forwardRef, HttpException, HttpStatus, Inject, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit, Optional } 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 { setTimeout as sleep } from 'node:timers/promises'; import { BillingService } from '../billing/billing.service'; import { isIpAllowed } from '../common/ip-allowlist'; import { moneyToNumber } from '../common/money'; import { PrismaService } from '../prisma/prisma.service'; import { RiskReviewService } from '../risk-review/risk-review.service'; import { OpenApiService } from '../open-api/open-api.service'; export interface CreateBatchTaskDto { tenantId: string; applicationId?: string; templateId?: string; content: string; category?: string; phones: string[]; sendMode?: 'immediate' | 'scheduled'; scheduledAt?: string; variables?: Record; createdById?: string; sourceIp?: string; userAgent?: string; sourceType?: 'client' | 'api' | 'cmpp'; clientMessageId?: string; } export type CreateHttpBatchTaskDto = Omit; export interface GatewayInboundAuthDto { account: string; password?: string; authSource?: string; timestamp?: number; remoteIp?: string; } export interface GatewayInboundSubmitDto { account: string; phoneNumber?: string; phoneNumbers?: string[]; content: string; srcId?: string; destId?: string; sequenceId?: number; remoteIp?: string; longMessage?: { reference: number; total: number; index: number; format: number; }; } interface GatewayInboundSingleSubmitResult { accepted: boolean; tenantId: string; applicationId: string; taskId: string; messageId: string; messageRecordId: string; status: 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; errorMessage?: 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' | 'unrecoverable' | 'queue_timeout'; type GatewayControlDeliveryResult = { sent?: boolean; delivered?: boolean; retryable?: boolean; reasonCode?: string; errorMessage?: string; 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; rawPayload?: string; deadLetteredAt?: string; } export interface RequeueGatewaySubmitExceptionDto { confirmedNotSubmitted?: boolean; reason?: string; operatorId?: 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 DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS = 72; const DEFAULT_RECEIPT_TIMEOUT_HOURS = 72; const DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS = 5 * 60_000; const RECEIPT_TIMEOUT_INITIAL_DELAY_MS = 60_000; const DEFAULT_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS = 5_000; const DEFAULT_SCHEDULED_DISPATCH_STALE_MS = 2 * 60_000; const SCHEDULED_DISPATCH_INITIAL_DELAY_MS = 1_000; const DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS = 2 * 60_000; const DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS = 2 * 60_000; const DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS = 60_000; const INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS = 10_000; const DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS = 30; const GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS = 30 * 24 * 60 * 60; const BULLMQ_PRIORITY: Record = { priority: 1, normal: 100, }; @Injectable() export class SendChainService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(SendChainService.name); private redis?: IORedis; private sendQueue?: Queue; private gatewayQueue?: Queue; private worker?: Worker; private receiptTimeoutInitialTimer?: ReturnType; private receiptTimeoutIntervalTimer?: ReturnType; private receiptTimeoutScanRunning = false; private scheduledDispatchInitialTimer?: ReturnType; private scheduledDispatchIntervalTimer?: ReturnType; private scheduledDispatchScanRunning = false; private inboundLongMessageInitialTimer?: ReturnType; private inboundLongMessageIntervalTimer?: ReturnType; constructor( private readonly prisma: PrismaService, private readonly billing: BillingService, private readonly riskReview: RiskReviewService, @Optional() @Inject(forwardRef(() => OpenApiService)) private readonly openApi?: OpenApiService, ) {} onModuleInit() { if (process.env.API_ENABLE_SEND_WORKER === 'true') { this.startWorker(); } if (process.env.SMS_RECEIPT_TIMEOUT_SCAN_ENABLED !== 'false') { this.receiptTimeoutInitialTimer = setTimeout(() => void this.runReceiptTimeoutScan(), RECEIPT_TIMEOUT_INITIAL_DELAY_MS); this.receiptTimeoutInitialTimer.unref?.(); this.receiptTimeoutIntervalTimer = setInterval( () => void this.runReceiptTimeoutScan(), positiveInteger(process.env.SMS_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS, DEFAULT_RECEIPT_TIMEOUT_SCAN_INTERVAL_MS), ); this.receiptTimeoutIntervalTimer.unref?.(); } if (process.env.SMS_SCHEDULED_DISPATCH_SCAN_ENABLED !== 'false') { this.scheduledDispatchInitialTimer = setTimeout( () => void this.runScheduledDispatchScan(), SCHEDULED_DISPATCH_INITIAL_DELAY_MS, ); this.scheduledDispatchInitialTimer.unref?.(); this.scheduledDispatchIntervalTimer = setInterval( () => void this.runScheduledDispatchScan(), positiveInteger(process.env.SMS_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS, DEFAULT_SCHEDULED_DISPATCH_SCAN_INTERVAL_MS), ); this.scheduledDispatchIntervalTimer.unref?.(); } if (process.env.CMPP_INBOUND_LONG_MESSAGE_SCAN_ENABLED !== 'false') { this.inboundLongMessageInitialTimer = setTimeout( () => void this.expireInboundLongMessages().catch((error) => { this.logger.error(`Failed to expire inbound CMPP long messages: ${String(error)}`); }), INBOUND_LONG_MESSAGE_SCAN_INITIAL_DELAY_MS, ); this.inboundLongMessageInitialTimer.unref?.(); this.inboundLongMessageIntervalTimer = setInterval( () => void this.expireInboundLongMessages().catch((error) => { this.logger.error(`Failed to expire inbound CMPP long messages: ${String(error)}`); }), positiveInteger( process.env.CMPP_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS, DEFAULT_INBOUND_LONG_MESSAGE_SCAN_INTERVAL_MS, ), ); this.inboundLongMessageIntervalTimer.unref?.(); } } async onModuleDestroy() { if (this.receiptTimeoutInitialTimer) clearTimeout(this.receiptTimeoutInitialTimer); if (this.receiptTimeoutIntervalTimer) clearInterval(this.receiptTimeoutIntervalTimer); if (this.scheduledDispatchInitialTimer) clearTimeout(this.scheduledDispatchInitialTimer); if (this.scheduledDispatchIntervalTimer) clearInterval(this.scheduledDispatchIntervalTimer); if (this.inboundLongMessageInitialTimer) clearTimeout(this.inboundLongMessageInitialTimer); if (this.inboundLongMessageIntervalTimer) clearInterval(this.inboundLongMessageIntervalTimer); 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 [messageClassification, unitPrice, queuePriority, accessNumber] = await Promise.all([ this.resolveTemplateMessageClassification(data.tenantId, data.applicationId, data.templateId, data.content), this.resolveUnitPrice(data.tenantId, data.applicationId), this.resolveQueuePriority(data.tenantId, data.applicationId), this.resolveApplicationAccessNumber(data.tenantId, data.applicationId), ]); const risk = messageClassification.rejectionReason ? { status: 'rejected', reason: messageClassification.rejectionReason, task: null } : await this.riskReview.evaluateTask({ tenantId: data.tenantId, applicationId: data.applicationId, templateId: data.templateId, content: data.content, category: data.category, phones, variables: messageClassification.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('企业账户余额不足'); } } if (data.applicationId && risk.status !== 'rejected') { await this.reserveDailySendQuota(data.applicationId, phones.length); } 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, signatureId: messageClassification.signatureId, drainageInfoId: messageClassification.drainageInfoId, messageId: `MSG-${randomUUID()}`, clientMessageId: data.clientMessageId, phoneNumber: phone, content: data.content, billingUnits: billing.billingUnitsPerMessage, unitPrice: billing.unitPrice, amountCents: billing.billingUnitsPerMessage * billing.unitPrice, queuePriority, clientSrcId: accessNumber.clientSrcId, applicationExtension: accessNumber.applicationExtension, 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, undefined, data.sourceType ?? 'client'); } async createHttpBatchTask(data: CreateHttpBatchTaskDto) { if (!data.applicationId) { throw new BadRequestException('公开 HTTP 发送必须关联企业应用'); } const template = await this.resolveInboundTemplateCandidate(data.applicationId, data.content); if (!template || template.auditStatus !== 'approved' || template.signature?.auditStatus !== 'approved') { throw new BadRequestException('短信内容未匹配当前应用已审核通过的签名和模板'); } const variables = matchTemplateContent(template.content, data.content); if (variables === null) { throw new BadRequestException('短信内容与已审核模板不匹配'); } return this.createBatchTask({ ...data, templateId: template.id, variables, sourceType: 'api', }); } 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(); 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(); 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 staleCutoff = new Date(now.getTime() - positiveInteger( process.env.SMS_SCHEDULED_DISPATCH_STALE_MS, DEFAULT_SCHEDULED_DISPATCH_STALE_MS, )); const tasks = await this.prisma.smsBatchTask.findMany({ where: { OR: [ { status: 'scheduled', scheduledAt: { lte: now } }, { status: { in: ['scheduled_dispatching', 'scheduled_recovering'] }, updatedAt: { lt: staleCutoff } }, ], }, orderBy: { scheduledAt: 'asc' }, }); const results: Array<{ taskId: string; status: string; enqueued?: number; reason?: string }> = []; for (const task of tasks) { const candidateStatus = task.status || 'scheduled'; const claimedStatus = candidateStatus === 'scheduled_dispatching' ? 'scheduled_recovering' : 'scheduled_dispatching'; const claimed = await this.prisma.smsBatchTask.updateMany({ where: { id: task.id, status: candidateStatus, ...(candidateStatus === 'scheduled' ? {} : { updatedAt: { lt: staleCutoff } }), }, data: { status: claimedStatus }, }); if (claimed.count !== 1) continue; let reservationEstablished = false; let dispatchPrepared = false; try { await this.validateSendResources(task.tenantId, task.applicationId ?? undefined, task.templateId ?? undefined); const messages = await this.prisma.smsMessageRecord.findMany({ where: { batchTaskId: task.id, status: { in: ['scheduled', 'queued'] } }, select: { id: true, amountCents: true, billingUnits: true }, take: 100000, }); const amountCents = messages.reduce((sum, message) => sum + moneyToNumber(message.amountCents), 0); const existingReservation = await this.prisma.accountTransaction.findFirst({ where: { tenantId: task.tenantId, transactionType: 'frozen', relatedType: 'sms_batch_task', relatedId: task.id }, select: { id: true }, }); reservationEstablished = Boolean(existingReservation); if (!reservationEstablished) { 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: '定时任务到点冻结', }); reservationEstablished = true; } } dispatchPrepared = true; 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 : '定时任务到点执行失败'; if (reservationEstablished || dispatchPrepared) { await this.prisma.smsBatchTask.update({ where: { id: task.id }, data: { status: claimedStatus, rejectReason: `调度将在超时后恢复:${reason}` }, }); results.push({ taskId: task.id, status: 'retrying', reason }); continue; } 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 }; } private async runScheduledDispatchScan() { if (this.scheduledDispatchScanRunning) return; this.scheduledDispatchScanRunning = true; try { await this.dispatchDueScheduledTasks(); } catch (error) { this.logger.error(`Scheduled SMS dispatch scan failed: ${error instanceof Error ? error.message : String(error)}`); } finally { this.scheduledDispatchScanRunning = false; } } startWorker() { if (this.worker) { return { status: 'already_started' }; } const connection = bullmqConnection(); this.worker = new Worker( 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: data.submitId ? { 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', 'requeueing', 'requeue_recovering', '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 logicalChannelId = resolved.channelId ?? data.channelId; const receiptKey = this.receiptEventKey(data, logicalChannelId); const existingReceipt = await this.prisma.smsReceiptRecord.findUnique({ where: { receiptKey }, include: { messageRecord: true }, }); if (existingReceipt?.messageRecord) { return existingReceipt.messageRecord; } const message = resolved.message; const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date(); if (resolved.submitRecordId) { await this.prisma.smsSubmitRecord.updateMany({ where: { id: resolved.submitRecordId, gatewayMessageId: null, }, data: { gatewayMessageId: data.gatewayMessageId, sequenceId: data.sequenceId, }, }); } try { await this.prisma.smsReceiptRecord.create({ data: { tenantId: message.tenantId, batchTaskId: message.batchTaskId, messageRecordId: message.id, receiptKey, channelId: logicalChannelId, messageId: resolved.messageId, gatewayMessageId: data.gatewayMessageId, phoneNumber: data.phoneNumber?.trim() || message.phoneNumber, sequenceId: data.sequenceId, receiptStatus: data.receiptStatus, rawStatus: data.rawStatus, errorCode: data.errorCode, errorMessage: data.errorMessage, deliveredAt, }, }); } catch (error) { if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === 'P2002') { const duplicate = await this.prisma.smsReceiptRecord.findUnique({ where: { receiptKey }, include: { messageRecord: true }, }); if (duplicate?.messageRecord) return duplicate.messageRecord; } throw error; } const logicalReceipt = { ...data, channelId: logicalChannelId }; await this.recordReceiptSegment(message, logicalReceipt, deliveredAt, resolved.submitRecordId); const aggregate = await this.aggregateReceiptSegments( message, logicalReceipt, deliveredAt, resolved.submitRecordId, resolved.submitId, ); if (!aggregate.terminal) { return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } }); } const status = aggregate.status; const isCurrentAttempt = (!message.channelId || message.channelId === logicalChannelId) && ( !message.gatewayMessageId || message.gatewayMessageId === data.gatewayMessageId || (aggregate.segmentTotal > 1 && (!message.submitId || message.submitId === resolved.submitId)) ); 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: { channelId: logicalChannelId, gatewayMessageId: message.gatewayMessageId ?? data.gatewayMessageId, receiptStatus: aggregate.receiptStatus, receiptRawStatus: aggregate.rawStatus, status, errorCode: aggregate.errorCode, errorMessage: aggregate.errorMessage ?? (status === 'delivered' ? null : aggregate.rawStatus), deliveredAt: aggregate.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: aggregate.receiptStatus, rawStatus: aggregate.rawStatus, errorCode: aggregate.errorCode, submitSequenceId: message.cmppSubmitSequenceId ? Number(message.cmppSubmitSequenceId) : undefined, submitGroupMessageId: message.cmppSubmitGroupMessageId ?? undefined, deliveredAt: aggregate.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(), uplinkMessageId: record.id, }, }); } 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; } if (failureType === 'queue_timeout' && delivery.status !== 'pending') { 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 nonRetryableFailure = failureType === 'unrecoverable' || failureType === 'queue_timeout'; const finalFailure = nonRetryableFailure || !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, failureCode: data.failureCode, failureMessage: data.failureMessage, attempts: data.attempts, maxAttempts: data.maxAttempts, commandPayload: data.commandPayload as Prisma.InputJsonValue | undefined, rawPayload: data.rawPayload, }, 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) => Promise; upsert: (args: Record) => Promise; }; }).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, data: RequeueGatewaySubmitExceptionDto = {}) { const deadLetter = await this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } }); if (!deadLetter) { throw new NotFoundException('Gateway提交异常记录不存在'); } if (deadLetter.status !== 'pending') { throw new BadRequestException('该提交异常当前状态不允许重新入队'); } if (!data.confirmedNotSubmitted) { throw new BadRequestException('请确认运营商未接收该短信后再重新入队'); } const reason = String(data.reason ?? '').trim(); if (reason.length < 5 || reason.length > 500) { throw new BadRequestException('请填写5至500字的重新入队原因'); } if (!deadLetter.commandPayload || !isObjectRecord(deadLetter.commandPayload)) { throw new BadRequestException('该提交异常缺少可重新入队的SubmitCommand'); } if (deadLetter.manualRetryCount >= 3) { throw new BadRequestException('该提交异常已达到人工重新入队次数上限'); } const message = deadLetter.messageId ? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: deadLetter.messageId } }) : null; if (message && ( message.submitStatus === 'accepted' || ['submitted', 'delivered', 'unknown'].includes(message.status) || ['delivered', 'unknown'].includes(message.receiptStatus ?? '') )) { throw new BadRequestException('该短信已有成功或不确定的上游结果,为避免重复发送,禁止重新入队'); } const commandChannelId = String(deadLetter.commandPayload.channelId ?? deadLetter.channelId ?? '').trim(); if (!commandChannelId) { throw new BadRequestException('该提交异常缺少通道信息'); } const channel = await this.prisma.smsChannel.findUnique({ where: { id: commandChannelId }, include: { connectionStates: true }, }); if (!channel || channel.status !== 'active') { throw new BadRequestException('原通道不存在或已停用,不能重新入队'); } if (!channel.connectionStates.some((state) => state.status === 'connected' && state.currentConnections > 0)) { throw new BadRequestException('原通道当前没有可用CMPP连接,请先恢复通道'); } const claimed = await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id, status: 'pending' }, data: { status: 'requeueing' }, }); if (claimed.count !== 1) { throw new BadRequestException('该提交异常已被其他操作处理,请刷新后重试'); } const requeueKey = gatewaySubmitRequeueKey(deadLetter.id, deadLetter.manualRetryCount + 1); let retryStreamMessageId: string; try { const publishedStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload, requeueKey); if (!publishedStreamMessageId) { throw new Error('Gateway提交异常重新入队未返回Stream消息编号'); } retryStreamMessageId = publishedStreamMessageId; } catch (error) { await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id, status: 'requeueing' }, data: { status: 'pending' }, }); throw error; } const finalized = await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id, status: 'requeueing' }, data: { status: 'requeued', manualRetryCount: { increment: 1 }, lastRetryStreamId: retryStreamMessageId, lastRetriedAt: new Date(), }, }); const updated = await this.prisma.gatewaySubmitDeadLetter.findUnique({ where: { id } }); if (!updated) { throw new NotFoundException('Gateway提交异常记录不存在'); } if (finalized.count !== 1 && updated.status !== 'resolved') { throw new BadRequestException('该提交异常状态已变化,请刷新后确认处理结果'); } await this.prisma.operationLog.create({ data: { tenantId: updated.tenantId ?? undefined, userId: data.operatorId, 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, reason, confirmedNotSubmitted: true, }, }, }); return updated; } async recoverStaleGatewaySubmitRequeues(now = new Date()) { const staleCutoff = new Date(now.getTime() - positiveInteger( process.env.GATEWAY_SUBMIT_REQUEUE_STALE_MS, DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS, )); const stale = await this.prisma.gatewaySubmitDeadLetter.findMany({ where: { status: 'requeueing', updatedAt: { lt: staleCutoff } }, orderBy: { updatedAt: 'asc' }, take: 100, }); let recovered = 0; let failed = 0; for (const deadLetter of stale) { if (!deadLetter.commandPayload || !isObjectRecord(deadLetter.commandPayload)) { await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id: deadLetter.id, status: 'requeueing', updatedAt: deadLetter.updatedAt }, data: { status: 'pending' }, }); failed += 1; continue; } const claimed = await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id: deadLetter.id, status: 'requeueing', updatedAt: deadLetter.updatedAt }, data: { status: 'requeue_recovering' }, }); if (claimed.count !== 1) continue; try { const requeueKey = gatewaySubmitRequeueKey(deadLetter.id, deadLetter.manualRetryCount + 1); const retryStreamMessageId = await this.publishGatewaySubmitCommand(deadLetter.commandPayload, requeueKey); if (!retryStreamMessageId) throw new Error('Gateway提交异常恢复未返回Stream消息编号'); const finalized = await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id: deadLetter.id, status: 'requeue_recovering' }, data: { status: 'requeued', manualRetryCount: { increment: 1 }, lastRetryStreamId: retryStreamMessageId, lastRetriedAt: new Date(), }, }); if (finalized.count === 1) { recovered += 1; await this.prisma.operationLog.create({ data: { tenantId: deadLetter.tenantId ?? undefined, action: 'gateway.submit_dead_letter_requeue_recovered', resource: 'gateway_submit_dead_letter', resourceId: deadLetter.id, detail: { retryStreamMessageId, requeueKey }, }, }); } } catch (error) { failed += 1; await this.prisma.gatewaySubmitDeadLetter.updateMany({ where: { id: deadLetter.id, status: 'requeue_recovering' }, data: { status: 'requeueing' }, }); this.logger.error(`Gateway submit requeue recovery failed for ${deadLetter.id}: ${error instanceof Error ? error.message : String(error)}`); } } return { recovered, failed }; } 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'); } if (delivery.status === 'awaiting_ack') { throw new BadRequestException('该记录正在等待客户端确认,不允许并发重投'); } 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}`); } const requestPayload = { deliveryId: delivery.id, account: String(payload.account ?? delivery.application?.cmppAccount ?? ''), ...payload, }; const retriedAt = new Date(); const claimed = await this.prisma.cmppDownstreamDelivery.updateMany({ where: { id: delivery.id, status: delivery.status, updatedAt: delivery.updatedAt, }, data: { status: 'manual_requeueing', retryCount: 0, manualRetryCount: { increment: 1 }, lastRetriedAt: retriedAt, nextRetryAt: null, sentAt: null, acknowledgedAt: null, ackDeadlineAt: null, ackResult: null, ackSequenceId: null, ackMessageId: null, connectionId: null, deliveredAt: null, lastError: null, }, }); if (claimed.count !== 1) { throw new BadRequestException('该下游投递记录已被其他操作处理,请刷新后重试'); } 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, previousStatus: delivery.status, previousRetryCount: delivery.retryCount, manualRetryCount: (delivery.manualRetryCount ?? 0) + 1, lastRetriedAt: retriedAt, }, }, }); try { const result = await this.postGatewayControl(path, requestPayload) as GatewayControlDeliveryResult; if (result.sent || result.delivered) { return this.markDownstreamDeliverySent({ id: delivery.id, ...result }); } return this.markDownstreamDeliveryFailed( delivery.id, downstreamControlFailureMessage(result), result.retryable === false ? 'unrecoverable' : 'send_failed', ); } catch (error) { return this.markDownstreamDeliveryFailed( delivery.id, error instanceof Error ? error.message : 'Gateway control delivery failed', ); } } async recoverStaleDownstreamManualRequeues(now = new Date()) { const staleCutoff = new Date(now.getTime() - positiveInteger( process.env.CMPP_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, )); const stale = await this.prisma.cmppDownstreamDelivery.findMany({ where: { status: 'manual_requeueing', updatedAt: { lt: staleCutoff } }, select: { id: true, updatedAt: true }, orderBy: { updatedAt: 'asc' }, take: 500, }); let recovered = 0; for (const delivery of stale) { const updated = await this.prisma.cmppDownstreamDelivery.updateMany({ where: { id: delivery.id, status: 'manual_requeueing', updatedAt: delivery.updatedAt }, data: { status: 'pending', nextRetryAt: null, lastError: '人工重投进程中断,已恢复为待投递', }, }); recovered += updated.count; } return { recovered }; } 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; }) { if (!data.applicationId) { return null; } const application = await this.prisma.smsApplication.findUnique({ where: { id: data.applicationId }, select: { cmppAccount: true, interfaceEnabled: true, downstreamReceiptRetryEnabled: true, downstreamUplinkRetryEnabled: true, httpConfig: true, }, }); try { await this.openApi?.queueWebhookEvent({ tenantId: data.tenantId, applicationId: data.applicationId, messageRecordId: data.messageRecordId, messageId: data.messageId, uplinkMessageId: typeof data.payload.uplinkMessageId === 'string' ? data.payload.uplinkMessageId : undefined, eventType: data.deliveryType, payload: data.payload, }); } catch (error) { this.logger.error(`HTTP webhook queue failed for ${data.deliveryType}/${data.messageId ?? '-'}: ${error instanceof Error ? error.message : String(error)}`); } if (application?.interfaceEnabled !== true) { return null; } 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 }); } else { await this.markDownstreamDeliveryFailed( delivery.id, downstreamControlFailureMessage(result), result.retryable === false ? 'unrecoverable' : 'send_failed', ); } } 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 && !isIpAllowed(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, maxConnections: application.cmppMaxConnections, status: 'authenticated', }; } async submitInboundMessage(data: GatewayInboundSubmitDto) { const phoneNumbers = data.phoneNumbers?.length ? data.phoneNumbers.map((phoneNumber) => phoneNumber.trim()) : data.phoneNumber ? [data.phoneNumber.trim()] : []; if (phoneNumbers.length === 0 || phoneNumbers.some((phoneNumber) => !/^1[3-9]\d{9}$/.test(phoneNumber))) { throw new BadRequestException('CMPP submit phone number is invalid'); } const application = await this.findInboundApplication(data.account); if (!application) { throw new BadRequestException('CMPP account is invalid'); } if (data.longMessage) { if (data.remoteIp && !isIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) { throw new BadRequestException('CMPP source IP is not in application allowlist'); } validateInboundApplicationSrcId(data.srcId, application); const collection = await this.collectInboundLongMessageFragment(data, application, phoneNumbers); if (collection.response) { return collection.response; } if (!collection.complete) { return { accepted: true, tenantId: application.tenantId, applicationId: application.id, messageId: collection.messageId, status: 'fragment_pending', fragmentPending: true, receivedSegments: collection.receivedSegments, segmentTotal: data.longMessage.total, phoneCount: phoneNumbers.length, messages: phoneNumbers.map((phoneNumber) => ({ phoneNumber, messageId: collection.messageId, status: 'fragment_pending', })), }; } try { const response = await this.recoverCompletedInboundLongMessageResponse( collection.messageId, phoneNumbers, ) ?? await this.submitCompleteInboundMessage({ ...data, content: collection.content, sequenceId: collection.sequenceId, longMessage: undefined, }, phoneNumbers, application, collection.messageId); await this.prisma.cmppInboundLongMessage.update({ where: { id: collection.groupId }, data: { status: 'completed', response: JSON.parse(JSON.stringify(response)) as Prisma.InputJsonValue, completedAt: new Date(), }, }); return response; } catch (error) { await this.prisma.cmppInboundLongMessage.update({ where: { id: collection.groupId }, data: { status: 'rejected', completedAt: new Date(), }, }).catch(() => undefined); throw error; } } return this.submitCompleteInboundMessage(data, phoneNumbers, application); } private async recoverCompletedInboundLongMessageResponse(messageId: string, phoneNumbers: string[]) { const existing = await this.prisma.smsMessageRecord.findMany({ where: { cmppSubmitGroupMessageId: messageId, phoneNumber: { in: phoneNumbers }, }, select: { id: true, tenantId: true, applicationId: true, batchTaskId: true, messageId: true, phoneNumber: true, status: true, errorCode: true, }, }); const byPhone = new Map(existing.map((item) => [item.phoneNumber, item])); const ordered = phoneNumbers.map((phoneNumber) => byPhone.get(phoneNumber)); if (ordered.some((item) => !item)) { return null; } const messages = ordered.map((item, index) => ({ phoneNumber: phoneNumbers[index], messageId: item!.messageId, messageRecordId: item!.id, taskId: item!.batchTaskId ?? '', status: item!.status, })); const first = ordered[0]!; const dailyLimitRejected = ordered.every((item) => item!.errorCode === 'DAILY_LIMIT'); return { accepted: !dailyLimitRejected, tenantId: first.tenantId ?? '', applicationId: first.applicationId ?? '', taskId: first.batchTaskId ?? '', messageId: first.messageId, messageRecordId: first.id, status: dailyLimitRejected ? 'rejected' : 'accepted', result: dailyLimitRejected ? 8 : undefined, phoneCount: messages.length, messages, }; } private async submitCompleteInboundMessage( data: GatewayInboundSubmitDto, phoneNumbers: string[], application: Awaited>, requestedGroupMessageId?: string, ) { if (!application) { throw new BadRequestException('CMPP account is invalid'); } const persisted = requestedGroupMessageId ? await this.prisma.smsMessageRecord.findMany({ where: { cmppSubmitGroupMessageId: requestedGroupMessageId, phoneNumber: { in: phoneNumbers }, }, select: { id: true, tenantId: true, applicationId: true, batchTaskId: true, messageId: true, phoneNumber: true, status: true, errorCode: true, }, }) : []; const persistedByPhone = new Map(persisted.map((item) => [item.phoneNumber, item])); const missingPhoneCount = phoneNumbers.filter((phoneNumber) => !persistedByPhone.has(phoneNumber)).length; const dailyQuota = missingPhoneCount > 0 ? await this.tryReserveDailySendQuota(application.id, missingPhoneCount) : { reserved: true, dailyLimit: application.dailyLimit ?? 100000 }; const dailyLimitRejection = dailyQuota.reserved ? undefined : { code: 'DAILY_LIMIT', reason: `应用当日发送上限${dailyQuota.dailyLimit}条,本次${missingPhoneCount}条超出剩余配额`, }; const submitGroupMessageId = requestedGroupMessageId ?? `MSG-${randomUUID()}`; const submissions = phoneNumbers.map((phoneNumber, index) => ({ phoneNumber, persisted: persistedByPhone.get(phoneNumber), messageId: persistedByPhone.get(phoneNumber)?.messageId ?? (index === 0 ? submitGroupMessageId : `MSG-${randomUUID()}`), })); const results: GatewayInboundSingleSubmitResult[] = []; const concurrency = 10; for (let offset = 0; offset < submissions.length; offset += concurrency) { const batch = submissions.slice(offset, offset + concurrency); results.push(...await Promise.all(batch.map((submission) => submission.persisted ? Promise.resolve({ accepted: submission.persisted.errorCode !== 'DAILY_LIMIT', tenantId: submission.persisted.tenantId ?? application.tenantId, applicationId: submission.persisted.applicationId ?? application.id, taskId: submission.persisted.batchTaskId ?? '', messageId: submission.persisted.messageId, messageRecordId: submission.persisted.id, status: submission.persisted.status, }) : this.submitInboundSingleMessage({ ...data, phoneNumber: submission.phoneNumber, phoneNumbers: undefined, }, submission.messageId, submitGroupMessageId, dailyLimitRejection)))); } const first = results[0]; return { ...first, result: dailyLimitRejection ? 8 : undefined, phoneCount: results.length, messages: results.map((result, index) => ({ phoneNumber: phoneNumbers[index], messageId: result.messageId, messageRecordId: result.messageRecordId, taskId: result.taskId, status: result.status, })), }; } private async collectInboundLongMessageFragment( data: GatewayInboundSubmitDto, application: NonNullable>>, phoneNumbers: string[], ) { const fragment = data.longMessage; if (!fragment || !Number.isInteger(fragment.reference) || fragment.reference < 0 || fragment.reference > 65535 || !Number.isInteger(fragment.total) || fragment.total < 2 || fragment.total > 255 || !Number.isInteger(fragment.index) || fragment.index < 1 || fragment.index > fragment.total || !Number.isInteger(fragment.format) || fragment.format < 0 || fragment.format > 255) { throw new BadRequestException('CMPP long message fragment metadata is invalid'); } const groupKey = createHash('sha256').update(JSON.stringify({ applicationId: application.id, account: data.account, srcId: data.srcId?.trim() ?? '', phoneNumbers, reference: fragment.reference, total: fragment.total, format: fragment.format, })).digest('hex'); const contentHash = createHash('sha256').update(data.content).digest('hex'); const now = new Date(); const expiresAt = new Date(now.getTime() + positiveInteger( process.env.CMPP_INBOUND_LONG_MESSAGE_TTL_SECONDS, 300, ) * 1000); return this.prisma.$transaction(async (tx) => { await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${groupKey}, 0))`; await tx.cmppInboundLongMessage.updateMany({ where: { groupKey, status: { in: ['collecting', 'processing'] }, expiresAt: { lte: now }, }, data: { status: 'expired', completedAt: now }, }); const recent = await tx.cmppInboundLongMessage.findFirst({ where: { groupKey, expiresAt: { gt: now }, }, include: { segments: { orderBy: { segmentIndex: 'asc' } } }, orderBy: { createdAt: 'desc' }, }); const matchingRecentSegment = recent?.segments.find((item) => item.segmentIndex === fragment.index); if (recent && ['completed', 'rejected'].includes(recent.status) && matchingRecentSegment?.contentHash === contentHash && matchingRecentSegment.sequenceId === (data.sequenceId == null ? null : String(data.sequenceId))) { return { complete: recent.status === 'completed', groupId: recent.id, messageId: recent.messageId, receivedSegments: recent.segments.length, response: recent.response as any, content: recent.segments.map((item) => item.content).join(''), sequenceId: parseOptionalSequenceId(recent.segments[0]?.sequenceId), }; } let group = recent && ['collecting', 'processing'].includes(recent.status) ? recent : null; if (!group) { group = await tx.cmppInboundLongMessage.create({ data: { tenantId: application.tenantId, applicationId: application.id, groupKey, account: data.account, srcId: data.srcId?.trim() || null, phoneNumbers, concatReference: fragment.reference, segmentTotal: fragment.total, msgFmt: fragment.format, messageId: `MSG-${randomUUID()}`, expiresAt, }, include: { segments: { orderBy: { segmentIndex: 'asc' } } }, }); } if (group.status === 'processing') { const processingStaleMs = positiveInteger( process.env.CMPP_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, DEFAULT_INBOUND_LONG_MESSAGE_PROCESSING_STALE_SECONDS, ) * 1000; const complete = group.segments.length === fragment.total && group.segments.every((item, index) => item.segmentIndex === index + 1); if (complete && now.getTime() - group.updatedAt.getTime() >= processingStaleMs) { await tx.cmppInboundLongMessage.update({ where: { id: group.id }, data: { status: 'processing', expiresAt }, }); return { complete: true, groupId: group.id, messageId: group.messageId, receivedSegments: group.segments.length, response: null, content: group.segments.map((item) => item.content).join(''), sequenceId: parseOptionalSequenceId(group.segments[0]?.sequenceId), }; } return { complete: false, groupId: group.id, messageId: group.messageId, receivedSegments: group.segments.length, response: group.response as any, content: '', sequenceId: undefined, }; } const existing = group.segments.find((item) => item.segmentIndex === fragment.index); if (existing && (existing.contentHash !== contentHash || existing.sequenceId !== (data.sequenceId == null ? null : String(data.sequenceId)))) { throw new BadRequestException(`CMPP long message fragment ${fragment.index} conflicts with the stored fragment`); } if (!existing) { await tx.cmppInboundLongMessageSegment.create({ data: { groupId: group.id, segmentIndex: fragment.index, sequenceId: data.sequenceId == null ? null : String(data.sequenceId), content: data.content, contentHash, }, }); } const segments = await tx.cmppInboundLongMessageSegment.findMany({ where: { groupId: group.id }, orderBy: { segmentIndex: 'asc' }, }); const complete = segments.length === fragment.total && segments.every((item, index) => item.segmentIndex === index + 1); if (complete) { await tx.cmppInboundLongMessage.update({ where: { id: group.id }, data: { status: 'processing', expiresAt }, }); } return { complete, groupId: group.id, messageId: group.messageId, receivedSegments: segments.length, response: null, content: complete ? segments.map((item) => item.content).join('') : '', sequenceId: parseOptionalSequenceId(segments[0]?.sequenceId), }; }); } async expireInboundLongMessages(now = new Date()) { return this.prisma.cmppInboundLongMessage.updateMany({ where: { status: { in: ['collecting', 'processing'] }, expiresAt: { lte: now }, }, data: { status: 'expired', completedAt: now, }, }); } private async submitInboundSingleMessage( data: GatewayInboundSubmitDto & { phoneNumber: string }, messageId: string, submitGroupMessageId: string, synchronousRejection?: { code: string; reason: string }, ) { const application = await this.findInboundApplication(data.account); if (!application) { throw new BadRequestException('CMPP account is invalid'); } if (data.remoteIp && !isIpAllowed(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 clientSrcId = validateInboundApplicationSrcId(data.srcId, application); const template = await this.resolveInboundTemplateCandidate(application.id, data.content); const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : {}; const unitPrice = moneyToNumber(application.customerUnitPrice); 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: synchronousRejection ? 'rejected' : 'validating', auditStatus: synchronousRejection ? 'rejected' : undefined, rejectReason: synchronousRejection?.reason, 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: synchronousRejection ? 'rejected' : 'accepted', }, }); const message = await this.prisma.smsMessageRecord.create({ data: { tenantId: application.tenantId, batchTaskId: task.id, applicationId: application.id, templateId: template?.id, messageId, phoneNumber: data.phoneNumber, content: data.content, billingUnits: billing.billingUnitsPerMessage, unitPrice: billing.unitPrice, amountCents: billing.amountCents, queuePriority, cmppSubmitSequenceId: data.sequenceId == null ? null : String(data.sequenceId), cmppSubmitGroupMessageId: submitGroupMessageId, clientSrcId, applicationExtension: application.cmppApplicationExtension, status: synchronousRejection ? 'rejected' : 'validating', errorCode: synchronousRejection?.code, errorMessage: synchronousRejection?.reason, }, }); if (synchronousRejection) { return { accepted: false, tenantId: application.tenantId, applicationId: application.id, taskId: task.id, messageId: message.messageId, messageRecordId: message.id, status: 'rejected', }; } 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); }; const queueAfterRiskChecks = async (options: { templateId?: string; signatureId?: string }) => { const drainage = await this.resolveDrainageInfoMatch(options.signatureId, data.content); const drainageInfoId = drainage?.id; const drainageReason = drainageRejectionReason(drainage); if (drainageReason) { await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { drainageInfoId, signatureId: options.signatureId }, }); await reject('DRAINAGE_NOT_APPROVED', drainageReason); return; } const risk = await this.riskReview.evaluateTask({ tenantId: application.tenantId, applicationId: application.id, templateId: options.templateId, content: data.content, variables: options.templateId ? templateVariables : undefined, phones: [data.phoneNumber], }); if (risk.status === 'rejected') { await reject('RISK', risk.reason || '短信被风控拒绝'); return; } if (risk.status === 'pending_review') { await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'pending_review', signatureId: options.signatureId, drainageInfoId }, }); await this.prisma.smsBatchTask.update({ where: { id: task.id }, data: { status: 'pending_review', riskTaskId: risk.task?.id, auditStatus: 'pending', reviewReason: risk.reason }, }); return; } const accountCheck = await this.billing.checkAccount({ tenantId: application.tenantId, amountCents: billing.amountCents, }); if (!accountCheck.canSend) { await reject('BALANCE', '企业账户余额不足'); return; } 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', signatureId: options.signatureId, drainageInfoId }, }); await this.prisma.smsBatchTask.update({ where: { id: task.id }, data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' }, }); await this.enqueueBatchTask(task.id); }; 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 drainage = await this.resolveDrainageInfoMatch(signature.id, data.content); const drainageReason = drainageRejectionReason(drainage); if (drainageReason) { await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { drainageInfoId: drainage?.id, signatureId: signature.id }, }); await reject('DRAINAGE_NOT_APPROVED', drainageReason); return { accepted: true, tenantId: application.tenantId, applicationId: application.id, messageId, messageRecordId: message.id, taskId: task.id, status: 'rejected', }; } 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, drainage?.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 && application.templateMismatchMode === 'direct_send') { const signature = await this.resolveInboundSignatureCandidate(application.id, data.content); if (!signature) { await reject('SIGNATURE', '短信内容未识别到已审核通过的签名'); } else { await queueAfterRiskChecks({ signatureId: signature.id }); } } 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 { await queueAfterRiskChecks({ templateId: template.id, signatureId: template.signature.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 ?? positiveInteger(process.env.SMS_RECEIPT_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS); const cutoff = new Date(Date.now() - olderThanHours * 60 * 60 * 1000); const candidates = await this.prisma.smsMessageRecord.findMany({ where: { tenantId: { not: null }, status: { in: ['submitted', 'unknown'] }, submittedAt: { lte: cutoff }, }, select: { id: true, tenantId: true, batchTaskId: true, messageId: true, amountCents: true, billingUnits: true }, take: 10000, }); const timedOutTaskIds = new Set(); let timeout = 0; for (const candidate of candidates) { if (!candidate.tenantId) continue; const transitioned = await this.prisma.smsMessageRecord.updateMany({ where: { id: candidate.id, status: { in: ['submitted', 'unknown'] } }, data: { status: 'timeout', timeoutAt: new Date(), errorMessage: `${olderThanHours}小时未收到明确回执,自动转超时` }, }); if (transitioned.count !== 1) continue; timeout += 1; await this.refundMessage(candidate as typeof candidate & { tenantId: string }, `${olderThanHours}小时未收到明确回执,自动超时退款`); if (candidate.batchTaskId) timedOutTaskIds.add(candidate.batchTaskId); } for (const batchTaskId of timedOutTaskIds) { await this.refreshTaskProgress(batchTaskId); } return { timeout }; } async markExpiredDownstreamDeliveries(olderThanHours = downstreamPendingTimeoutHours()) { const cutoff = new Date(Date.now() - olderThanHours * 60 * 60_000); const expired = await this.prisma.cmppDownstreamDelivery.findMany({ where: { status: 'pending', OR: [ { lastRetriedAt: null, createdAt: { lte: cutoff } }, { lastRetriedAt: { lte: cutoff } }, ], }, select: { id: true }, take: 500, }); for (const delivery of expired) { await this.markDownstreamDeliveryFailed( delivery.id, `下游投递排队超过 ${olderThanHours} 小时,系统自动终止重试`, 'queue_timeout', ); } return { failed: expired.length }; } private async runReceiptTimeoutScan() { if (this.receiptTimeoutScanRunning) return; this.receiptTimeoutScanRunning = true; try { const [receiptResult, downstreamResult, requeueRecoveryResult, downstreamManualRecoveryResult] = await Promise.all([ this.markUnknownTimeout({}), this.markExpiredDownstreamDeliveries(), this.recoverStaleGatewaySubmitRequeues(), this.recoverStaleDownstreamManualRequeues(), ]); if (receiptResult.timeout > 0) this.logger.log(`Marked ${receiptResult.timeout} messages as receipt timeout and refunded charged messages`); if (downstreamResult.failed > 0) this.logger.log(`Terminated ${downstreamResult.failed} expired downstream deliveries`); if (requeueRecoveryResult.recovered > 0) this.logger.log(`Recovered ${requeueRecoveryResult.recovered} stale Gateway submit requeues`); if (downstreamManualRecoveryResult.recovered > 0) this.logger.log(`Recovered ${downstreamManualRecoveryResult.recovered} stale downstream manual requeues`); } catch (error) { this.logger.error('Receipt timeout scan failed', error instanceof Error ? error.stack : String(error)); } finally { this.receiptTimeoutScanRunning = false; } } 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; clientSrcId?: string | null; applicationExtension?: 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; const upstreamSrcId = composeUpstreamSrcId(channel.srcId, message.applicationExtension); 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', costUnitPrice: channel.unitPrice ?? 0, costAmountCents: moneyToNumber(channel.unitPrice) * Math.max(1, message.billingUnits ?? 1), }, }); 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: upstreamSrcId, 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), heartbeatIntervalSeconds: getPositiveConfigInteger(channel.config, 'heartbeatIntervalSeconds', 30), heartbeatMissThreshold: getPositiveConfigInteger(channel.config, 'heartbeatMissThreshold', 3), }, 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; clientSrcId?: string | null; applicationExtension?: string | null; }, 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 { 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, unitPrice: moneyToNumber(selected.channel.unitPrice) }, 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 moneyToNumber(application.customerUnitPrice); } private async resolveQueuePriority(tenantId: string, applicationId?: string): Promise { 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 async resolveApplicationAccessNumber(tenantId: string, applicationId?: string) { if (!applicationId) { return { clientSrcId: null, applicationExtension: null }; } const application = await this.prisma.smsApplication.findUnique({ where: { id: applicationId }, select: { tenantId: true, cmppClientSrcId: true, cmppApplicationExtension: true }, }); if (!application || application.tenantId !== tenantId) { return { clientSrcId: null, applicationExtension: null }; } return { clientSrcId: application.cmppClientSrcId, applicationExtension: application.cmppApplicationExtension, }; } private findInboundApplication(account: string) { return this.prisma.smsApplication.findFirst({ where: { cmppAccount: account }, include: { tenant: true, ipAllowlist: true, }, }); } private async resolveInboundTemplateCandidate(applicationId: string, content: string) { const exact = await this.prisma.smsTemplate.findFirst({ where: { applicationId, content, auditStatus: 'approved', signature: { auditStatus: 'approved' }, }, include: { signature: true }, orderBy: { updatedAt: 'desc' }, }); if (exact) return exact; const variableTemplates = await this.prisma.smsTemplate.findMany({ where: { applicationId, content: { contains: '${' }, auditStatus: 'approved', signature: { auditStatus: 'approved' }, }, include: { signature: true }, orderBy: { updatedAt: 'desc' }, }); return variableTemplates.find((template) => matchTemplateContent(template.content, content) !== null) ?? null; } 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 resolveTemplateMessageClassification( tenantId: string, applicationId: string | undefined, templateId: string | undefined, content: string, ) { if (templateId) { const template = await this.prisma.smsTemplate.findUnique({ where: { id: templateId }, include: { signature: true }, }); if (!template || template.tenantId !== tenantId || template.applicationId !== applicationId || template.auditStatus !== 'approved' || template.signature?.auditStatus !== 'approved') { throw new BadRequestException('短信模板不存在、未通过审核或不属于当前应用'); } const variables = matchTemplateContent(template.content, content); if (variables === null) { throw new BadRequestException('短信内容与选定的审核模板不匹配'); } const drainage = await this.resolveDrainageInfoMatch(template.signatureId, content); return { signatureId: template.signatureId, drainageInfoId: drainage?.id, variables, rejectionReason: drainageRejectionReason(drainage), }; } if (!applicationId) { throw new BadRequestException('自由内容短信必须关联企业应用'); } const [application, signature] = await Promise.all([ this.prisma.smsApplication.findUnique({ where: { id: applicationId }, select: { tenantId: true, templateMismatchMode: true }, }), this.resolveInboundSignatureCandidate(applicationId, content), ]); if (!application || application.tenantId !== tenantId) { throw new BadRequestException('短信应用不存在或不属于当前企业'); } if (!signature) { throw new BadRequestException('短信内容未以当前应用已审核通过的签名开头'); } if (application.templateMismatchMode !== 'direct_send') { throw new BadRequestException('当前应用未允许无模板自由内容直接发送'); } const drainage = await this.resolveDrainageInfoMatch(signature.id, content); return { signatureId: signature.id, drainageInfoId: drainage?.id, variables: undefined, rejectionReason: drainageRejectionReason(drainage), }; } private async resolveDrainageInfoMatch(signatureId: string | null | undefined, content: string) { if (!signatureId) return undefined; const candidates = await this.prisma.smsDrainageInfo.findMany({ where: { signatureId, auditStatus: { not: 'deleted' } }, select: { id: true, url: true, auditStatus: true, updatedAt: true }, orderBy: [{ updatedAt: 'desc' }, { id: 'asc' }], }); const matches = candidates .map((item) => ({ ...item, normalizedUrl: item.url.trim() })) .filter((item) => item.normalizedUrl.length > 0 && content.includes(item.normalizedUrl)) .sort((left, right) => right.normalizedUrl.length - left.normalizedUrl.length || right.updatedAt.getTime() - left.updatedAt.getTime()); if (matches.length === 0) return undefined; const longestLength = matches[0].normalizedUrl.length; const longestMatches = matches.filter((item) => item.normalizedUrl.length === longestLength); if (longestMatches.length !== 1) { throw new BadRequestException({ code: 'DRAINAGE_MATCH_AMBIGUOUS', message: '短信内容同时匹配多条等长引流地址,无法确定报备资料', drainageInfoIds: longestMatches.map((item) => item.id), }); } const matched = longestMatches[0]; return { id: matched.id, auditStatus: matched.auditStatus }; } private async attachMessageToReviewTask(reviewTaskId: string, messageRecordId: string, signatureId: string, drainageInfoId?: string) { await this.prisma.smsMessageRecord.update({ where: { id: messageRecordId }, data: { reviewTaskId, signatureId, drainageInfoId, 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; cmppSubmitGroupMessageId?: 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', receiptRawStatus: 'REJECTD', errorCode, errorMessage: reason, deliveredAt }, }); const gatewayMessageId = `PLATFORM:${message.messageId}`; const receipt = await this.prisma.smsReceiptRecord.create({ data: { tenantId: message.tenantId, batchTaskId: message.batchTaskId, messageRecordId: message.id, receiptKey: createHash('sha256').update(`platform\u0000${gatewayMessageId}\u0000${message.phoneNumber}\u0000undelivered\u0000REJECTD\u0000${errorCode}`).digest('hex'), messageId: message.messageId, gatewayMessageId, phoneNumber: message.phoneNumber, receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode, errorMessage: reason, 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, submitGroupMessageId: message.cmppSubmitGroupMessageId ?? 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 reserveDailySendQuota(applicationId: string, requestedCount: number) { const result = await this.tryReserveDailySendQuota(applicationId, requestedCount); if (!result.reserved) { throw new HttpException({ code: 'DAILY_SEND_LIMIT_EXCEEDED', message: `应用当日发送上限${result.dailyLimit}条,本次${requestedCount}条超出剩余配额`, dailyLimit: result.dailyLimit, requestedCount, }, HttpStatus.TOO_MANY_REQUESTS); } return result; } private async tryReserveDailySendQuota(applicationId: string, requestedCount: number) { if (!Number.isInteger(requestedCount) || requestedCount <= 0) { throw new BadRequestException('发送号码数量必须为正整数'); } const usageDate = shanghaiDateKey(); const reservationId = randomUUID(); const rows = await this.prisma.$queryRaw>(Prisma.sql` WITH application_limit AS ( SELECT id, COALESCE("dailyLimit", 100000)::integer AS "dailyLimit" FROM "SmsApplication" WHERE id = ${applicationId} ), reservation AS ( INSERT INTO "SmsApplicationDailyUsage" ( id, "applicationId", "usageDate", "usedCount", "createdAt", "updatedAt" ) SELECT ${reservationId}, id, ${usageDate}::date, ${requestedCount}, NOW(), NOW() FROM application_limit WHERE ${requestedCount} <= "dailyLimit" ON CONFLICT ("applicationId", "usageDate") DO UPDATE SET "usedCount" = "SmsApplicationDailyUsage"."usedCount" + EXCLUDED."usedCount", "updatedAt" = NOW() WHERE "SmsApplicationDailyUsage"."usedCount" + EXCLUDED."usedCount" <= (SELECT "dailyLimit" FROM application_limit) RETURNING "usedCount" ) SELECT application_limit."dailyLimit", reservation."usedCount" FROM application_limit LEFT JOIN reservation ON TRUE `); if (rows.length === 0) { throw new NotFoundException('短信应用不存在'); } return { dailyLimit: Number(rows[0].dailyLimit), usedCount: rows[0].usedCount == null ? null : Number(rows[0].usedCount), reserved: rows[0].usedCount != null, }; } private async chargeAcceptedMessage(message: { tenantId: string; applicationId?: string | null; batchTaskId: string; messageId: string; phoneNumber: string; content: string; billingUnits: number; unitPrice: number | bigint; amountCents: number | bigint; }) { const amountCents = moneyToNumber(message.amountCents); const unitPrice = moneyToNumber(message.unitPrice); 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, 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 | bigint; billingUnits: number }, remark: string, ) { const amountCents = moneyToNumber(message.amountCents); if (amountCents <= 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, relatedType: 'sms_message_record', relatedId: message.messageId, remark: `${remark}: ${message.messageId}`, }); } private async refundMessage( message: { tenantId: string; messageId: string; amountCents: number | bigint; billingUnits: number }, remark: string, ) { const amountCents = moneyToNumber(message.amountCents); if (amountCents <= 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, 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) => Promise; updateMany: (args: Record) => Promise<{ count: number }>; findFirst: (args: Record) => Promise; findMany: (args: Record) => Promise; }; }).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 aggregateReceiptSegments( message: { id: string; billingUnits?: number | null; }, data: GatewayReceiptEventDto, deliveredAt: Date, submitRecordId?: string, submitId?: string, ) { const audits = await this.smsMessageSegmentAuditDelegate().findMany({ where: submitRecordId ? { messageRecordId: message.id, submitRecordId } : submitId ? { messageRecordId: message.id, submitId } : { messageRecordId: message.id, gatewayMessageId: data.gatewayMessageId }, orderBy: { segmentIndex: 'asc' }, }); if (audits.length === 0) { const status = data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed'; return { terminal: true, segmentTotal: 1, status, receiptStatus: data.receiptStatus, rawStatus: data.rawStatus, errorCode: data.errorCode, errorMessage: data.errorMessage, deliveredAt, }; } const segmentTotal = Math.max( 1, Number(message.billingUnits ?? 1), ...audits.map((audit) => Number(audit.segmentTotal ?? 1)), ); const received = audits.filter((audit) => Boolean(audit.receiptStatus)); const failed = received.find((audit) => !['delivered', 'unknown'].includes(audit.receiptStatus ?? '')); if (failed) { return { terminal: true, segmentTotal, status: 'failed', receiptStatus: failed.receiptStatus ?? 'undelivered', rawStatus: failed.rawStatus ?? data.rawStatus, errorCode: failed.errorCode ?? data.errorCode, errorMessage: failed.errorMessage ?? data.errorMessage, deliveredAt: failed.deliveredAt ?? deliveredAt, }; } const delivered = received.filter((audit) => audit.receiptStatus === 'delivered'); if (delivered.length >= segmentTotal) { const latest = delivered.reduce((current, audit) => (audit.deliveredAt?.getTime() ?? 0) > (current.deliveredAt?.getTime() ?? 0) ? audit : current); return { terminal: true, segmentTotal, status: 'delivered', receiptStatus: 'delivered', rawStatus: latest.rawStatus ?? data.rawStatus, errorCode: latest.errorCode ?? undefined, errorMessage: undefined, deliveredAt: latest.deliveredAt ?? deliveredAt, }; } if (received.length >= segmentTotal) { const latest = received[received.length - 1]; return { terminal: true, segmentTotal, status: 'unknown', receiptStatus: 'unknown', rawStatus: latest.rawStatus ?? data.rawStatus, errorCode: latest.errorCode ?? data.errorCode, errorMessage: latest.errorMessage ?? data.errorMessage, deliveredAt: latest.deliveredAt ?? deliveredAt, }; } return { terminal: false, segmentTotal, status: 'submitted', receiptStatus: data.receiptStatus, rawStatus: data.rawStatus, errorCode: data.errorCode, errorMessage: data.errorMessage, 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 = data.messageId ? await this.prisma.smsMessageRecord.findUnique({ where: { messageId: data.messageId } }) : null; if (exactMessage) { const segmentAudit = data.gatewayMessageId ? await this.smsMessageSegmentAuditDelegate().findFirst({ where: { messageRecordId: exactMessage.id, gatewayMessageId: data.gatewayMessageId, }, orderBy: { updatedAt: 'desc' }, }) : null; if (segmentAudit) { return { message: exactMessage, messageId: exactMessage.messageId, submitRecordId: segmentAudit.submitRecordId ?? undefined, submitId: segmentAudit.submitId, channelId: segmentAudit.channelId ?? data.channelId, }; } const submitRecord = await this.prisma.smsSubmitRecord.findFirst({ where: { messageRecordId: exactMessage.id, channelId: data.channelId, gatewayMessageId: data.gatewayMessageId, }, orderBy: { createdAt: 'desc' }, }); return { message: exactMessage, messageId: exactMessage.messageId, submitRecordId: submitRecord?.id, submitId: submitRecord?.submitId, channelId: submitRecord?.channelId ?? data.channelId, }; } const phoneNumber = data.phoneNumber?.trim(); const exactSubmits = await this.prisma.smsSubmitRecord.findMany({ where: { channelId: data.channelId, gatewayMessageId: data.gatewayMessageId, ...(phoneNumber ? { messageRecord: { phoneNumber } } : {}), }, include: { messageRecord: true }, orderBy: { createdAt: 'desc' }, take: 2, }); if (exactSubmits.length === 1 && exactSubmits[0]?.messageRecord) { return { message: exactSubmits[0].messageRecord, messageId: exactSubmits[0].messageRecord.messageId, submitRecordId: exactSubmits[0].id, submitId: exactSubmits[0].submitId, channelId: exactSubmits[0].channelId, }; } if (!phoneNumber) { throw new NotFoundException('SMS message record not found'); } const incomingChannel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } }); if (!incomingChannel) { throw new NotFoundException('SMS message record not found'); } const segmentMatches = await this.smsMessageSegmentAuditDelegate().findMany({ where: { gatewayMessageId: data.gatewayMessageId, messageRecord: { phoneNumber }, }, include: { messageRecord: true, submitRecord: true, channel: true }, orderBy: { createdAt: 'desc' }, take: 10, }); const exactSegmentMatches = segmentMatches.filter((candidate) => candidate.channelId === data.channelId); if (exactSegmentMatches.length === 1 && exactSegmentMatches[0]?.messageRecord) { return { message: exactSegmentMatches[0].messageRecord, messageId: exactSegmentMatches[0].messageRecord.messageId, submitRecordId: exactSegmentMatches[0].submitRecordId ?? undefined, submitId: exactSegmentMatches[0].submitRecord?.submitId ?? exactSegmentMatches[0].submitId, channelId: exactSegmentMatches[0].channelId, }; } const sameSupplierSegments = segmentMatches.filter((candidate) => candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); if (sameSupplierSegments.length === 1 && sameSupplierSegments[0]?.messageRecord) { return { message: sameSupplierSegments[0].messageRecord, messageId: sameSupplierSegments[0].messageRecord.messageId, submitRecordId: sameSupplierSegments[0].submitRecordId ?? undefined, submitId: sameSupplierSegments[0].submitRecord?.submitId ?? sameSupplierSegments[0].submitId, channelId: sameSupplierSegments[0].channelId, }; } const crossConnectionSubmits = await this.prisma.smsSubmitRecord.findMany({ where: { gatewayMessageId: data.gatewayMessageId, messageRecord: { phoneNumber }, }, include: { messageRecord: true, channel: true }, orderBy: { createdAt: 'desc' }, take: 10, }); const sameSupplierSubmits = crossConnectionSubmits.filter((candidate) => candidate.channel && this.isSameSupplierConnection(incomingChannel, candidate.channel)); if (sameSupplierSubmits.length === 1 && sameSupplierSubmits[0]?.messageRecord) { return { message: sameSupplierSubmits[0].messageRecord, messageId: sameSupplierSubmits[0].messageRecord.messageId, submitRecordId: sameSupplierSubmits[0].id, submitId: sameSupplierSubmits[0].submitId, channelId: sameSupplierSubmits[0].channelId, }; } 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, submitId: candidates[0].submitId, channelId: candidates[0].channelId, }; } private isSameSupplierConnection( left: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, right: { account: string; gatewayHost: string; gatewayPort: number; protocol: string; cmppVersion: string }, ) { return left.account.trim() === right.account.trim() && left.gatewayHost.trim().toLowerCase() === right.gatewayHost.trim().toLowerCase() && left.gatewayPort === right.gatewayPort && left.protocol.trim().toUpperCase() === right.protocol.trim().toUpperCase() && left.cmppVersion.trim() === right.cmppVersion.trim(); } private receiptEventKey(data: GatewayReceiptEventDto, channelId = data.channelId) { return createHash('sha256').update([ channelId, data.gatewayMessageId, data.phoneNumber?.trim() ?? '', data.receiptStatus, data.rawStatus.trim(), data.errorCode ?? '', ].join('\u0000')).digest('hex'); } private getSendQueue(): Queue { if (!this.sendQueue) { this.sendQueue = new Queue(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, idempotencyKey?: string) { const redis = this.getRedis(); const stream = process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM; const payload = JSON.stringify(command); if (!idempotencyKey) { return redis.xadd(stream, '*', 'messageType', 'SubmitCommand', 'data', payload); } const result = await redis.eval( `local existing = redis.call('GET', KEYS[2]) if existing then return existing end local streamId = redis.call('XADD', KEYS[1], '*', 'messageType', 'SubmitCommand', 'data', ARGV[1]) redis.call('SET', KEYS[2], streamId, 'EX', ARGV[2]) return streamId`, 2, stream, idempotencyKey, payload, String(GATEWAY_SUBMIT_REQUEUE_IDEMPOTENCY_TTL_SECONDS), ); return typeof result === 'string' ? result : String(result ?? ''); } } function gatewaySubmitRequeueKey(deadLetterId: string, attempt: number) { return `gateway:submit:requeue:${deadLetterId}:${attempt}`; } function drainageRejectionReason(drainage?: { id: string; auditStatus: string }) { if (!drainage || drainage.auditStatus === 'approved') return undefined; return `短信内容匹配的引流资料 ${drainage.id} 当前为 ${drainage.auditStatus},必须审核通过后才能发送`; } 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 { 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 downstreamPendingTimeoutHours() { const value = Number(process.env.CMPP_DOWNSTREAM_PENDING_TIMEOUT_HOURS ?? DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS); return Number.isFinite(value) && value > 0 ? value : DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS; } function downstreamControlFailureMessage(result: GatewayControlDeliveryResult) { const reason = String(result.errorMessage ?? '').trim(); const code = String(result.reasonCode ?? '').trim(); if (reason && code) return `${reason} (${code})`; if (reason) return reason; if (code) return `Gateway 未完成下游投递 (${code})`; return 'Gateway 未完成下游投递,等待自动重试'; } 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 } = { 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)[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)[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 matchTemplateContent(templateContent: string, actualContent: string) { if (templateContent === actualContent) { return {} as Record; } const tokenPattern = /\$\{([a-zA-Z0-9_]+)\}/g; const names: string[] = []; let cursor = 0; let pattern = '^'; for (const match of templateContent.matchAll(tokenPattern)) { const index = match.index ?? 0; pattern += escapeRegularExpression(templateContent.slice(cursor, index)); pattern += '([\\s\\S]+?)'; names.push(match[1]); cursor = index + match[0].length; } if (names.length === 0) { return null; } pattern += `${escapeRegularExpression(templateContent.slice(cursor))}$`; const matched = new RegExp(pattern, 'u').exec(actualContent); if (!matched) { return null; } const variables: Record = {}; for (let index = 0; index < names.length; index += 1) { const name = names[index]; const value = matched[index + 1]; if (variables[name] !== undefined && variables[name] !== value) { return null; } variables[name] = value; } return variables; } function escapeRegularExpression(value: string) { return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); } 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 validateInboundApplicationSrcId( srcId: string | undefined, application: { cmppApplicationExtension?: string | null; cmppAccessNumberFillEnabled?: boolean | null; cmppAccessNumberFillPrefix?: string | null; cmppClientSrcId?: string | null; }, ) { const submittedSrcId = srcId?.trim() ?? ''; const applicationExtension = application.cmppApplicationExtension?.trim() ?? ''; if (!applicationExtension) { return submittedSrcId || null; } const fillPrefix = application.cmppAccessNumberFillEnabled ? application.cmppAccessNumberFillPrefix?.trim() ?? '' : ''; const expectedSrcId = application.cmppClientSrcId?.trim() || `${fillPrefix}${applicationExtension}`; if (!submittedSrcId || submittedSrcId !== expectedSrcId) { throw new BadRequestException(`CMPP Src_Id must equal the access number assigned to this application: ${expectedSrcId}`); } return submittedSrcId; } function composeUpstreamSrcId(baseSrcId: string, applicationExtension?: string | null) { const upstreamSrcId = `${baseSrcId.trim()}${applicationExtension?.trim() ?? ''}`; if (upstreamSrcId.length > 21) { throw new BadRequestException('channel base access number plus application extension must not exceed 21 digits'); } return upstreamSrcId; } function positiveInteger(value: string | undefined, fallback: number) { const parsed = Number(value); return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; } function parseOptionalSequenceId(value: string | null | undefined) { if (!value) return undefined; const parsed = Number(value); return Number.isInteger(parsed) && parsed >= 0 && parsed <= 0xffffffff ? parsed : undefined; } function shanghaiDateKey(now = new Date()) { const parts = new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai', year: 'numeric', month: '2-digit', day: '2-digit', }).formatToParts(now); const values = Object.fromEntries(parts.map((part) => [part.type, part.value])); return `${values.year}-${values.month}-${values.day}`; } 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 | null, current: Record, ) { 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; }