360 lines
16 KiB
TypeScript
360 lines
16 KiB
TypeScript
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';
|
|
|
|
|
|
/**
|
|
* R10 retry implementation.
|
|
* Cross-domain calls return through the stable SendChainService seam to preserve locking and test observability.
|
|
*/
|
|
export class SendRetryService {
|
|
private readonly logger = new Logger('SendChainService');
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly billing: BillingService,
|
|
private readonly openApi: OpenApiService | undefined,
|
|
private readonly facade: SendCompletionFacade,
|
|
private readonly callbacks: SendCompletionCallbacks,
|
|
) {}
|
|
|
|
async 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 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.facade.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.facade.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 retryMessageIfAllowed(
|
|
message: {
|
|
id: string;
|
|
tenantId: string;
|
|
batchTaskId: string;
|
|
applicationId?: string | null;
|
|
templateId?: string | null;
|
|
signatureId?: string | null;
|
|
submitId?: string | null;
|
|
messageId: string;
|
|
phoneNumber: string;
|
|
content: string;
|
|
billingUnits: number;
|
|
queuedAt?: Date;
|
|
clientSrcId?: string | null;
|
|
applicationExtension?: string | null;
|
|
carrier?: string | null;
|
|
province?: string | null;
|
|
},
|
|
reason: string,
|
|
sourceSubmitRecordId?: string,
|
|
) {
|
|
const attempts = await this.prisma.smsSubmitRecord.findMany({
|
|
where: { messageRecordId: message.id },
|
|
orderBy: { createdAt: 'asc' },
|
|
take: 200,
|
|
});
|
|
const attemptedChannelIds = attempts.map((attempt) => attempt.channelId);
|
|
let sourceAttempt = sourceSubmitRecordId
|
|
? attempts.find((attempt) => attempt.id === sourceSubmitRecordId)
|
|
: attempts.find((attempt) => attempt.submitId === message.submitId) ?? attempts[attempts.length - 1];
|
|
if (!sourceAttempt && sourceSubmitRecordId) {
|
|
sourceAttempt = await this.prisma.smsSubmitRecord.findUnique({
|
|
where: { id: sourceSubmitRecordId },
|
|
}) ?? undefined;
|
|
}
|
|
if (!sourceAttempt || sourceAttempt.messageRecordId !== message.id) {
|
|
this.logger.error(`sms_retry_route_failed ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
reason,
|
|
sourceSubmitRecordId,
|
|
sourceMessageRecordId: sourceAttempt?.messageRecordId,
|
|
error: sourceAttempt ? 'retry_source_submit_record_mismatch' : 'retry_source_submit_record_missing',
|
|
})}`);
|
|
return null;
|
|
}
|
|
const existingRetry = await this.prisma.smsSubmitRecord.findUnique({
|
|
where: { retryOfSubmitRecordId: sourceAttempt.id },
|
|
});
|
|
if (existingRetry) {
|
|
this.logger.warn(`sms_retry_claim_reused ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
retryOfSubmitRecordId: sourceAttempt.id,
|
|
submitId: existingRetry.submitId,
|
|
channelId: existingRetry.channelId,
|
|
})}`);
|
|
return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
|
|
}
|
|
const ageMinutes = (Date.now() - new Date(message.queuedAt ?? Date.now()).getTime()) / 60_000;
|
|
this.logger.log(`sms_retry_route_started ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
reason,
|
|
attemptedChannelIds,
|
|
ageMinutes: Math.round(ageMinutes * 100) / 100,
|
|
})}`);
|
|
if (ageMinutes >= 72 * 60) {
|
|
this.logger.warn(`sms_retry_route_skipped ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
reason: 'maximum_message_age_exceeded',
|
|
ageMinutes: Math.round(ageMinutes * 100) / 100,
|
|
})}`);
|
|
return null;
|
|
}
|
|
const retryCarrier = message.carrier
|
|
? normalizeCarrier(message.carrier)
|
|
: await this.facade.identifyCarrier(message.phoneNumber);
|
|
const route = await this.facade.findApplicationRoute(message.tenantId, message.applicationId ?? undefined, retryCarrier);
|
|
const retryTimeLimitMinutes = Math.min(route.group.retryTimeLimitMinutes ?? route.group.retryTimeLimitHours * 60, 72 * 60);
|
|
if (!route.group.retryEnabled || ageMinutes >= retryTimeLimitMinutes) {
|
|
this.logger.warn(`sms_retry_route_skipped ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
groupId: route.groupId,
|
|
reason: !route.group.retryEnabled ? 'group_retry_disabled' : 'group_retry_time_limit_exceeded',
|
|
ageMinutes: Math.round(ageMinutes * 100) / 100,
|
|
retryTimeLimitMinutes,
|
|
})}`);
|
|
return null;
|
|
}
|
|
try {
|
|
const routed = await this.facade.selectChannelForMessage({ ...message, carrier: retryCarrier }, {
|
|
forceNational: true,
|
|
excludeChannelIds: attemptedChannelIds,
|
|
});
|
|
await this.prisma.smsMessageRecord.update({
|
|
where: { id: message.id },
|
|
data: { errorMessage: reason },
|
|
});
|
|
const retried = await this.facade.submitMessageToGateway(
|
|
message,
|
|
routed,
|
|
attempts.length,
|
|
sourceAttempt.id,
|
|
);
|
|
this.logger.log(`sms_retry_route_selected ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
groupId: routed.groupId,
|
|
channelId: routed.channel.id,
|
|
attempt: attempts.length,
|
|
})}`);
|
|
return retried;
|
|
} catch (error) {
|
|
this.logger.error(`sms_retry_route_failed ${JSON.stringify({
|
|
messageId: message.messageId,
|
|
messageRecordId: message.id,
|
|
reason,
|
|
attemptedChannelIds,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
})}`);
|
|
return null;
|
|
}
|
|
}
|
|
}
|