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

fix: harden admin and CMPP delivery workflows

Browse Source
This commit is contained in:
hectorzhao
2026-07-15 11:21:02 +08:00
parent 00b6d95752
commit e47432bc9d
34 changed files with 878 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.ts
+135 -57
View File
@@ -1263,6 +1263,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
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');
@@ -1277,30 +1280,19 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
throw new BadRequestException(`Unsupported downstream delivery type ${delivery.deliveryType}`);
}
await this.prisma.operationLog.create({
data: {
tenantId: delivery.tenantId,
action: 'gateway.downstream_delivery_requeue',
resource: 'cmpp_downstream_delivery',
resourceId: delivery.id,
detail: {
deliveryType: delivery.deliveryType,
applicationId: delivery.applicationId,
messageId: delivery.messageId,
},
},
});
const requestPayload = {
deliveryId: delivery.id,
account: String(payload.account ?? delivery.application?.cmppAccount ?? ''),
...payload,
};
await this.prisma.cmppDownstreamDelivery.update({
const retriedAt = new Date();
const requeued = await this.prisma.cmppDownstreamDelivery.update({
where: { id: delivery.id },
data: {
status: 'pending',
retryCount: 0,
manualRetryCount: { increment: 1 },
lastRetriedAt: retriedAt,
nextRetryAt: null,
sentAt: null,
acknowledgedAt: null,
@@ -1313,6 +1305,23 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
lastError: null,
},
});
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: requeued.manualRetryCount,
lastRetriedAt: retriedAt,
},
},
});
try {
const result = await this.postGatewayControl(path, requestPayload) as GatewayControlDeliveryResult;
if (result.sent || result.delivered) {
@@ -1649,6 +1658,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
throw new BadRequestException('CMPP submit phone number is invalid');
}
const template = await this.resolveInboundTemplateCandidate(application.id, data.content);
const templateVariables = template ? matchTemplateContent(template.content, data.content) ?? {} : {};
const unitPrice = application.customerUnitPrice ?? 0;
const queuePriority = normalizeQueuePriority(application.queuePriority);
const billing = this.billing.estimateSmsCost({
@@ -1707,6 +1717,57 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
await this.recordCmppFailureReceipt(message, code, reason);
};
const queueAfterRiskChecks = async (options: { templateId?: string; signatureId?: string }) => {
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 },
});
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 },
});
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) {
@@ -1765,6 +1826,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
}
}
} 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') {
@@ -1772,46 +1840,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
} else if (!template.signature || template.signature.auditStatus !== 'approved') {
await reject('SIGNATURE', '短信签名尚未审核通过');
} else {
const risk = await this.riskReview.evaluateTask({
tenantId: application.tenantId,
applicationId: application.id,
templateId: template.id,
content: data.content,
phones: [data.phoneNumber],
});
if (risk.status === 'rejected') {
await reject('RISK', risk.reason || '短信被风控拒绝');
} else if (risk.status === 'pending_review') {
await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'pending_review' } });
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'pending_review', riskTaskId: risk.task?.id, auditStatus: 'pending', reviewReason: risk.reason },
});
} else {
const accountCheck = await this.billing.checkAccount({
tenantId: application.tenantId,
amountCents: billing.amountCents,
});
if (!accountCheck.canSend) {
await reject('BALANCE', '企业账户余额不足');
} else {
if (billing.amountCents > 0) {
await this.billing.freeze({
tenantId: application.tenantId,
amountCents: billing.amountCents,
relatedType: 'sms_batch_task',
relatedId: task.id,
remark: 'CMPP 入站短信冻结',
});
}
await this.prisma.smsMessageRecord.update({ where: { id: message.id }, data: { status: 'queued' } });
await this.prisma.smsBatchTask.update({
where: { id: task.id },
data: { status: 'ready', riskTaskId: risk.task?.id, auditStatus: 'approved' },
});
await this.enqueueBatchTask(task.id);
}
}
await queueAfterRiskChecks({ templateId: template.id, signatureId: template.signature.id });
}
return {
accepted: true,
@@ -2160,8 +2189,8 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
private resolveInboundTemplateCandidate(applicationId: string, content: string) {
return this.prisma.smsTemplate.findFirst({
private async resolveInboundTemplateCandidate(applicationId: string, content: string) {
const exact = await this.prisma.smsTemplate.findFirst({
where: {
applicationId,
content,
@@ -2169,6 +2198,16 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
include: { signature: true },
orderBy: { updatedAt: 'desc' },
});
if (exact) return exact;
const variableTemplates = await this.prisma.smsTemplate.findMany({
where: {
applicationId,
content: { contains: '${' },
},
include: { signature: true },
orderBy: { updatedAt: 'desc' },
});
return variableTemplates.find((template) => matchTemplateContent(template.content, content) !== null) ?? null;
}
private resolveInboundSignatureCandidate(applicationId: string, content: string) {
@@ -2902,6 +2941,45 @@ function normalizeRegion(region?: string | null) {
return String(region ?? '').replace(/省|市|自治区|壮族|回族|维吾尔/g, '').trim();
}
function matchTemplateContent(templateContent: string, actualContent: string) {
if (templateContent === actualContent) {
return {} as Record<string, string>;
}
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<string, string> = {};
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);