import { BadRequestException, HttpException, HttpStatus, Logger, NotFoundException } from '@nestjs/common'; import { Prisma } from '@prisma/client'; import { createHash } from 'node:crypto'; import { BillingService } from '../billing/billing.service'; import { moneyToNumber } from '../common/money'; import type { OpenApiService } from '../open-api/open-api.service'; import { PrismaService } from '../prisma/prisma.service'; import type { GatewaySubmitResultDto, GatewaySubmitSegmentResultDto, GatewayReceiptEventDto, GatewayUplinkEventDto, GatewayPendingDeliveryQueryDto, GatewayDownstreamSentDto, GatewayDownstreamAcknowledgedDto, GatewayDownstreamFailureType, GatewayControlDeliveryResult, GatewaySubmitDeadLetterDto, RequeueGatewaySubmitExceptionDto, GatewayDownstreamRecoveryStatusDto, TimeoutUnknownDto } from './send-chain.contracts'; import { positiveInteger, normalizeReceiptStatus, normalizeCarrier, DEFAULT_DOWNSTREAM_RETRY_DELAY_MS, DEFAULT_DOWNSTREAM_RETRY_MAX_DELAY_MS, DEFAULT_DOWNSTREAM_MAX_RETRIES, DEFAULT_DOWNSTREAM_PENDING_TIMEOUT_HOURS, DEFAULT_RECEIPT_TIMEOUT_HOURS, DEFAULT_GATEWAY_SUBMIT_REQUEUE_STALE_MS, DEFAULT_DOWNSTREAM_MANUAL_REQUEUE_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_PROCESSING_STALE_MS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_ATTEMPTS, DEFAULT_UPSTREAM_RECEIPT_INBOX_MAX_AGE_HOURS, gatewaySubmitRequeueKey, isObjectRecord, asDateOrNull, downstreamRetryDelayMs, downstreamAckTimeoutMs, downstreamRetryBaseDelayMs, downstreamRetryMaxDelayMs, downstreamMaxRetries, downstreamPendingTimeoutHours, downstreamControlFailureMessage, normalizeSubmitStatus, downstreamDeliveryAttemptKey, hasRecoveryAuditStateChanged, normalizeRecoveryFailureCategory, aggregateReceiptSegmentState, isSameUpstreamEndpointIdentity, receiptEventKey } from './send-chain.helpers'; import type { SendSubmissionService } from './send-submission.service'; import type { SendCompletionCallbacks, SendCompletionFacade } from './send-completion.service'; import { queueFinalReceiptDeliveries } from './downstream-receipt-targets'; /** * R10 timeout implementation. * Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability. */ export class SendTimeoutService { private readonly logger = new Logger('SendChainService'); private receiptTimeoutScanRunning = false; constructor( private readonly prisma: PrismaService, private readonly billing: BillingService, private readonly openApi: OpenApiService | undefined, private readonly facade: SendCompletionFacade, private readonly callbacks: SendCompletionCallbacks, ) {} 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 }, OR: [ { status: { in: ['submitted', 'unknown'] }, submittedAt: { lte: cutoff } }, { status: 'timeout', errorCode: 'RECEIPT_TIMEOUT', timeoutReceiptQueuedAt: null }, ], }, select: { id: true, tenantId: true, batchTaskId: true, applicationId: true, messageId: true, phoneNumber: true, amountCents: true, billingUnits: true, status: true, cmppSubmitSequenceId: true, cmppSubmitGroupMessageId: true, cmppRegisteredDelivery: true, timeoutAt: true, }, take: 10000, }); const timedOutTaskIds = new Set(); let timeout = 0; for (const candidate of candidates) { if (!candidate.tenantId) continue; const timedOutAt = candidate.timeoutAt ?? new Date(); if (candidate.status !== 'timeout') { const transitioned = await this.prisma.smsMessageRecord.updateMany({ where: { id: candidate.id, status: { in: ['submitted', 'unknown'] } }, data: { status: 'timeout', receiptStatus: 'undelivered', receiptRawStatus: 'EXPIRED', errorCode: 'RECEIPT_TIMEOUT', errorMessage: `${olderThanHours}小时未收到明确回执,自动转超时`, timeoutAt: timedOutAt, }, }); if (transitioned.count !== 1) continue; timeout += 1; } // Refund uses the platform-message idempotency key. Re-running it for a // timeout whose downstream outbox was not fully queued also recovers a // crash between the state transition and the original refund call. await this.facade.refundMessage(candidate as typeof candidate & { tenantId: string }, `${olderThanHours}小时未收到明确回执,自动超时退款`); const queued = await queueFinalReceiptDeliveries( this.prisma, (request) => this.facade.queueAndTryDownstreamDelivery(request), { message: candidate, payload: { messageId: candidate.messageId, gatewayMessageId: `PLATFORM_TIMEOUT:${candidate.messageId}`, phoneNumber: candidate.phoneNumber, receiptStatus: 'undelivered', rawStatus: 'EXPIRED', errorCode: 'RECEIPT_TIMEOUT', errorMessage: `${olderThanHours}小时未收到明确回执,自动转超时`, deliveredAt: timedOutAt.toISOString(), }, propagateHttpQueueError: true, }, ); if (queued.queued) { await this.prisma.smsMessageRecord.updateMany({ where: { id: candidate.id, status: 'timeout', timeoutReceiptQueuedAt: null }, data: { timeoutReceiptQueuedAt: new Date() }, }); } if (candidate.batchTaskId) timedOutTaskIds.add(candidate.batchTaskId); } for (const batchTaskId of timedOutTaskIds) { await this.facade.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.facade.markDownstreamDeliveryFailed( delivery.id, `下游投递排队超过 ${olderThanHours} 小时,系统自动终止重试`, 'queue_timeout', ); } return { failed: expired.length }; } async runReceiptTimeoutScan() { if (this.receiptTimeoutScanRunning) return; this.receiptTimeoutScanRunning = true; try { const [receiptResult, downstreamResult, requeueRecoveryResult, downstreamManualRecoveryResult] = await Promise.all([ this.facade.markUnknownTimeout({}), this.facade.markExpiredDownstreamDeliveries(), this.facade.recoverStaleGatewaySubmitRequeues(), this.facade.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; } } }