hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: record cmpp business failures with receipts

Browse Source
This commit is contained in:
hectorzhao
2026-07-11 18:30:57 +08:00
parent 2ecbb920a6
commit 76e0417ccf
6 changed files with 260 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+51 -3
View File
@@ -106,6 +106,7 @@ function createPrismaMock() {
findMany: jest.fn(),
},
smsMessageRecord: {
create: jest.fn().mockImplementation(({ data }) => Promise.resolve({ ...message, ...data })),
createMany: jest.fn().mockResolvedValue({ count: 2 }),
findMany: jest.fn().mockResolvedValue([{ id: 'record-1', batchTaskId: 'task-1' }]),
findUnique: jest.fn().mockResolvedValue(message),
@@ -149,6 +150,7 @@ function createPrismaMock() {
},
smsReceiptRecord: {
create: jest.fn().mockResolvedValue({ id: 'receipt-1' }),
findFirst: jest.fn().mockResolvedValue(null),
findMany: jest.fn(),
},
smsUplinkMessage: {
@@ -474,7 +476,7 @@ describe('SendChainService', () => {
})).rejects.toThrow('CMPP interface is disabled for this application');
});
it('rejects Gateway submit when application interface is disabled', async () => {
it('records and acknowledges Gateway submit with a failure receipt when the application interface was disabled after bind', async () => {
const { service, prisma } = createService();
prisma.smsApplication.findFirst.mockResolvedValue({
id: 'app-1',
@@ -493,9 +495,35 @@ describe('SendChainService', () => {
phoneNumber: '13800000001',
content: 'hello',
remoteIp: '127.0.0.1',
})).rejects.toThrow('CMPP interface is disabled for this application');
})).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' }));
expect(prisma.smsBatchTask.create).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.create).toHaveBeenCalledWith({ data: expect.objectContaining({ status: 'validating' }) });
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ messageRecordId: 'record-1', receiptStatus: 'undelivered', errorCode: 'INTERFACE' }),
});
expect(prisma.cmppDownstreamDelivery.create).toHaveBeenCalledWith({
data: expect.objectContaining({ messageRecordId: 'record-1', deliveryType: 'receipt', status: 'pending' }),
});
});
it('records an unreported CMPP message and returns success before delivering the template failure receipt', async () => {
const { service, prisma } = createService();
prisma.smsTemplate.findFirst.mockResolvedValue(null);
await expect(service.submitInboundMessage({
account: '100001',
phoneNumber: '13800000001',
content: 'unreported content',
remoteIp: '127.0.0.1',
})).resolves.toEqual(expect.objectContaining({ accepted: true, messageRecordId: 'record-1' }));
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }),
});
expect(service['postGatewayControl']).toHaveBeenCalledWith(
'/downstream/receipt',
expect.objectContaining({ receiptStatus: 'undelivered', rawStatus: 'REJECTD', errorCode: 'TEMPLATE' }),
);
});
it('previews imported phone files with duplicate, invalid, blacklist, and variable errors', async () => {
@@ -896,11 +924,31 @@ describe('SendChainService', () => {
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
prisma.channelSignatureReportTask.findFirst.mockResolvedValue(null);
prisma.smsMessageRecord.findUnique.mockResolvedValue({
id: 'record-1',
tenantId: 'tenant-1',
batchTaskId: 'task-1',
applicationId: 'app-1',
templateId: 'tpl-1',
messageId: 'MSG-1',
phoneNumber: '13800000001',
content: 'hello',
billingUnits: 1,
unitPrice: 3,
amountCents: 3,
status: 'queued',
queuePriority: 'normal',
batchTask: { sourceType: 'cmpp' },
template: { signature: { id: 'sig-1', name: '签名' } },
});
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
expect.objectContaining({ submitted: false, status: 'failed', reason: '短信签名未在最终通道报备通过' }),
);
expect(gatewayAdd).not.toHaveBeenCalled();
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ receiptStatus: 'undelivered', errorCode: 'ROUTE' }),
});
});
it('only selects channel group items allocated to the matched carrier', async () => {
api/src/send-chain/send-chain.service.ts
+191 -29
View File
@@ -626,13 +626,20 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
data: { status: 'failed', errorMessage: reason },
});
await this.releaseMessageReservation(businessMessage, reason);
await this.refreshTaskProgress(businessMessage.batchTaskId);
if (message.batchTask?.sourceType === 'cmpp') {
await this.recordCmppFailureReceipt(businessMessage, 'ROUTE', reason);
} else {
await this.refreshTaskProgress(businessMessage.batchTaskId);
}
return { submitted: false, messageRecordId: message.id, status: 'failed', reason };
}
}
async handleSubmitResult(data: GatewaySubmitResultDto) {
const message = await this.requireMessageByGatewayEvent(data.messageId, data.gatewayMessageId);
const batchTask = message.batchTaskId
? await this.prisma.smsBatchTask.findUnique({ where: { id: message.batchTaskId }, select: { sourceType: true } })
: null;
const submittedAt = data.submittedAt ? new Date(data.submittedAt) : new Date();
await this.prisma.smsSubmitRecord.updateMany({
where: { OR: [{ submitId: data.submitId }, { messageRecordId: message.id }] },
@@ -675,6 +682,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
timeoutAt: data.submitStatus === 'timeout' ? submittedAt : undefined,
},
});
if (data.submitStatus !== 'accepted' && batchTask?.sourceType === 'cmpp' && message.tenantId && message.applicationId) {
await this.recordCmppFailureReceipt(
message as typeof message & { tenantId: string; applicationId: string; batchTaskId: string },
data.errorCode || 'SUBMIT',
data.errorMessage || (data.submitStatus === 'timeout' ? '上游提交超时' : '上游拒绝短信'),
);
}
await this.prisma.gatewaySubmitDeadLetter.updateMany({
where: {
status: { in: ['pending', 'requeued'] },
@@ -1422,11 +1436,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
async submitInboundMessage(data: GatewayInboundSubmitDto) {
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) {
throw new BadRequestException('CMPP account is invalid');
}
if (data.remoteIp && !isApplicationIpAllowed(data.remoteIp, application.ipAllowlist.map((item) => item.ipCidr))) {
throw new BadRequestException('CMPP source IP is not in application allowlist');
@@ -1434,26 +1445,130 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (!/^1[3-9]\d{9}$/.test(data.phoneNumber)) {
throw new BadRequestException('CMPP submit phone number is invalid');
}
const template = await this.resolveInboundTemplate(application.id, data.content);
const task = await this.createBatchTask({
const template = await this.resolveInboundTemplateCandidate(application.id, data.content);
const unitPrice = application.customerUnitPrice ?? 0;
const queuePriority = normalizeQueuePriority(application.queuePriority);
const billing = this.billing.estimateSmsCost({
tenantId: application.tenantId,
applicationId: application.id,
templateId: template.id,
content: data.content,
phones: [data.phoneNumber],
sourceType: 'cmpp',
sourceIp: data.remoteIp,
userAgent: 'cmpp-gateway',
phoneCount: 1,
unitPrice,
});
const message = task?.messages?.[0];
const task = await this.prisma.smsBatchTask.create({
data: {
tenantId: application.tenantId,
applicationId: application.id,
templateId: template?.id,
taskNo: `BT-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceType: 'cmpp',
content: data.content,
phoneTotal: 1,
status: 'validating',
progressTotal: 1,
},
});
await this.prisma.smsApiRequest.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
requestId: `REQ-${Date.now()}-${randomUUID().slice(0, 8)}`,
sourceIp: data.remoteIp,
userAgent: 'cmpp-gateway',
payloadSummary: { phoneTotal: 1, contentLength: [...data.content].length, account: data.account },
status: 'accepted',
},
});
const message = await this.prisma.smsMessageRecord.create({
data: {
tenantId: application.tenantId,
batchTaskId: task.id,
applicationId: application.id,
templateId: template?.id,
messageId: `MSG-${randomUUID()}`,
phoneNumber: data.phoneNumber,
content: data.content,
billingUnits: billing.billingUnitsPerMessage,
unitPrice: billing.unitPrice,
amountCents: billing.amountCents,
queuePriority,
status: 'validating',
},
});
const reject = async (code: string, reason: string) => {
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'rejected', auditStatus: 'rejected', rejectReason: reason },
});
await this.recordCmppFailureReceipt(message, code, reason);
};
if (application.status !== 'active' || application.tenant.status !== 'active') {
await reject('ACCOUNT', '企业或短信应用已停用');
} else if (!application.interfaceEnabled) {
await reject('INTERFACE', '短信应用 CMPP 接口已停用');
} else if (application.tenant.certificationStatus !== 'approved') {
await reject('CERT', '企业认证未通过');
} else if (!template) {
await reject('TEMPLATE', '短信内容未匹配到已报备模板');
} else if (template.auditStatus !== 'approved') {
await reject('TEMPLATE', '短信模板尚未审核通过');
} else if (!template.signature || template.signature.auditStatus !== 'approved') {
await reject('SIGNATURE', '短信签名尚未审核通过');
} else if (template.signature.reportStatus !== 'approved') {
await reject('REPORT', '短信签名尚未报备通过');
} else {
const risk = await this.riskReview.evaluateTask({
tenantId: application.tenantId,
applicationId: application.id,
templateId: template.id,
content: data.content,
phones: [data.phoneNumber],
});
if (risk.status === 'rejected') {
await reject('RISK', risk.reason || '短信被风控拒绝');
} else if (risk.status === 'pending_review') {
await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'pending_review' } });
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'pending_review', riskTaskId: risk.task?.id, auditStatus: 'pending', reviewReason: risk.reason },
});
} else {
const accountCheck = await this.billing.checkAccount({
tenantId: application.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
});
if (!accountCheck.canSend) {
await reject('BALANCE', '企业账户余额、套餐余量或授信额度不足');
} else {
if (billing.amountCents + billing.totalBillingUnits > 0) {
await this.billing.freeze({
tenantId: application.tenantId,
amountCents: billing.amountCents,
smsUnits: billing.totalBillingUnits,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: 'CMPP 入站短信冻结',
});
}
await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'queued' } });
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' },
});
await this.enqueueBatchTask(task.id);
}
}
}
return {
accepted: true,
tenantId: application.tenantId,
applicationId: application.id,
taskId: task?.id,
messageId: message?.messageId,
messageRecordId: message?.id,
status: message?.status ?? task?.status,
taskId: task.id,
messageId: message.messageId,
messageRecordId: message.id,
status: 'accepted',
};
}
@@ -1762,24 +1877,71 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
private async resolveInboundTemplate(applicationId: string, content: string) {
const template = await this.prisma.smsTemplate.findFirst({
private resolveInboundTemplateCandidate(applicationId: string, content: string) {
return this.prisma.smsTemplate.findFirst({
where: {
applicationId,
content,
auditStatus: 'approved',
signature: {
auditStatus: 'approved',
reportStatus: 'approved',
},
},
include: { signature: true },
orderBy: { updatedAt: 'desc' },
});
if (!template) {
throw new BadRequestException('CMPP submit content does not match an approved template and signature');
}
return template;
}
private async recordCmppFailureReceipt(
message: {
id: string;
tenantId?: string | null;
batchTaskId?: string | null;
applicationId?: string | null;
messageId: string;
phoneNumber: string;
},
errorCode: string,
reason: string,
) {
if (!message.tenantId || !message.applicationId) return null;
const existing = await this.prisma.smsReceiptRecord.findFirst({
where: { messageRecordId: message.id, gatewayMessageId: `PLATFORM:${message.messageId}` },
});
if (existing) return existing;
const deliveredAt = new Date();
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { status: 'failed', receiptStatus: 'undelivered', errorCode, errorMessage: reason, deliveredAt },
});
const receipt = await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
messageId: message.messageId,
gatewayMessageId: `PLATFORM:${message.messageId}`,
receiptStatus: 'undelivered',
rawStatus: 'REJECTD',
errorCode,
deliveredAt,
},
});
await this.queueAndTryDownstreamDelivery({
tenantId: message.tenantId,
applicationId: message.applicationId,
messageRecordId: message.id,
messageId: message.messageId,
deliveryType: 'receipt',
payload: {
messageId: message.messageId,
gatewayMessageId: `PLATFORM:${message.messageId}`,
phoneNumber: message.phoneNumber,
receiptStatus: 'undelivered',
rawStatus: 'REJECTD',
errorCode,
errorMessage: reason,
deliveredAt: deliveredAt.toISOString(),
},
});
if (message.batchTaskId) await this.refreshTaskProgress(message.batchTaskId);
return receipt;
}
private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) {