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

fix: enforce carrier-specific channel group routing

Browse Source
This commit is contained in:
hectorzhao
2026-07-03 12:48:16 +08:00
parent a890b03163
commit dd09d91c1e
17 changed files with 630 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+139 -4
View File
@@ -17,7 +17,10 @@ function createPrismaMock() {
unitPrice: 3,
amountCents: 3,
status: 'queued',
template: { signature: { name: '签名' } },
submitId: 'SUB-1',
gatewayMessageId: 'GW-1',
channelId: 'channel-1',
template: { signature: { id: 'sig-1', name: '签名' } },
};
const channel = {
id: 'channel-1',
@@ -42,10 +45,11 @@ function createPrismaMock() {
status: 'active',
group: {
id: 'group-1',
carrier: 'mobile',
status: 'active',
retryEnabled: true,
retryTimeLimitHours: 72,
items: [{ id: 'item-1', groupId: 'group-1', channelId: 'channel-1', priority: 1, province: null, channel }],
items: [{ id: 'item-1', groupId: 'group-1', channelId: 'channel-1', carrier: 'mobile', priority: 1, province: null, channel }],
},
};
return {
@@ -53,7 +57,7 @@ function createPrismaMock() {
findUnique: jest.fn().mockResolvedValue({ id: 'tenant-1', status: 'active', certificationStatus: 'approved' }),
},
smsApplication: {
findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active' }),
findUnique: jest.fn().mockResolvedValue({ id: 'app-1', tenantId: 'tenant-1', status: 'active', customerUnitPrice: 3 }),
},
smsTemplate: {
findUnique: jest.fn().mockResolvedValue({
@@ -104,6 +108,9 @@ function createPrismaMock() {
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
findMany: jest.fn().mockResolvedValue([]),
},
channelSignatureReportTask: {
findFirst: jest.fn().mockResolvedValue({ id: 'report-task-1' }),
},
smsReceiptRecord: {
create: jest.fn().mockResolvedValue({ id: 'receipt-1' }),
findMany: jest.fn(),
@@ -118,6 +125,9 @@ function createPrismaMock() {
update: jest.fn().mockResolvedValue({ id: 'bill-1' }),
updateMany: jest.fn().mockResolvedValue({ count: 1 }),
},
accountTransaction: {
findFirst: jest.fn().mockResolvedValue(null),
},
enterpriseBlacklist: {
findMany: jest.fn().mockResolvedValue([]),
},
@@ -342,13 +352,17 @@ describe('SendChainService', () => {
it('releases reservation for rejected submit result and refunds failed receipts', async () => {
const { service, prisma, billing } = createService();
prisma.smsBillingRecord.findFirst
.mockResolvedValueOnce(null)
.mockResolvedValueOnce(null)
.mockResolvedValueOnce({ id: 'bill-1', billingStatus: 'charged' });
prisma.channelRouteRule.findFirst.mockResolvedValue({
id: 'route-1',
tenantId: 'tenant-1',
applicationId: 'app-1',
groupId: 'group-1',
carrier: 'mobile',
group: { id: 'group-1', status: 'active', retryEnabled: false, retryTimeLimitHours: 72, items: [] },
group: { id: 'group-1', carrier: 'mobile', status: 'active', retryEnabled: false, retryTimeLimitHours: 72, items: [] },
});
await service.handleSubmitResult({
@@ -369,6 +383,127 @@ describe('SendChainService', () => {
expect(billing.refund).toHaveBeenCalledWith(expect.objectContaining({ remark: '最终失败退款' }));
});
it('does not let stale failed receipts overwrite a later delivered message', async () => {
const { service, prisma, billing } = createService();
prisma.smsMessageRecord.findFirst.mockResolvedValue({
id: 'record-1',
tenantId: 'tenant-1',
batchTaskId: 'task-1',
applicationId: 'app-1',
templateId: 'tpl-1',
messageId: 'MSG-1',
phoneNumber: '13800000001',
content: 'hello',
channelId: 'channel-new',
gatewayMessageId: 'GW-NEW',
status: 'delivered',
amountCents: 3,
billingUnits: 1,
unitPrice: 3,
});
await service.handleReceipt({
messageId: 'MSG-1',
channelId: 'channel-old',
gatewayMessageId: 'GW-OLD',
receiptStatus: 'undelivered',
rawStatus: 'UNDELIV',
});
expect(prisma.smsReceiptRecord.create).toHaveBeenCalledWith({
data: expect.objectContaining({ channelId: 'channel-old', gatewayMessageId: 'GW-OLD', receiptStatus: 'undelivered' }),
});
expect(billing.refund).not.toHaveBeenCalled();
expect(prisma.smsMessageRecord.update).not.toHaveBeenCalledWith({
where: { id: 'record-1' },
data: expect.objectContaining({ status: 'failed' }),
});
});
it('blocks submit when signature is not approved on the selected channel', async () => {
const { service, prisma } = createService();
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
prisma.channelSignatureReportTask.findFirst.mockResolvedValue(null);
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
expect.objectContaining({ submitted: false, status: 'failed', reason: '短信签名未在最终通道报备通过' }),
);
expect(gatewayAdd).not.toHaveBeenCalled();
});
it('only selects channel group items allocated to the matched carrier', async () => {
const { service, prisma } = createService();
const gatewayAdd = jest.fn().mockResolvedValue(undefined);
service['waitForChannelRateLimit'] = jest.fn().mockResolvedValue(undefined);
service['getGatewayQueue'] = jest.fn().mockReturnValue({ add: gatewayAdd });
prisma.channelRouteRule.findFirst.mockResolvedValue({
id: 'route-1',
tenantId: 'tenant-1',
applicationId: 'app-1',
groupId: 'group-1',
carrier: 'mobile',
group: {
id: 'group-1',
carrier: 'mobile',
status: 'active',
retryEnabled: true,
retryTimeLimitHours: 72,
items: [
{
id: 'wrong-item',
groupId: 'group-1',
channelId: 'channel-unicom',
carrier: 'unicom',
priority: 1,
province: null,
channel: {
id: 'channel-unicom',
code: 'CMPP-U',
account: 'u',
srcId: '1061',
rateLimitPerSecond: 100,
unitPrice: 3,
status: 'active',
carrier: 'all',
sendRegion: '全国',
connectionStates: [{ status: 'online', currentConnections: 1, desiredConnections: 1 }],
},
},
{
id: 'mobile-item',
groupId: 'group-1',
channelId: 'channel-all',
carrier: 'mobile',
priority: 2,
province: null,
channel: {
id: 'channel-all',
code: 'CMPP-ALL',
account: 'all',
srcId: '1062',
rateLimitPerSecond: 100,
unitPrice: 3,
status: 'active',
carrier: 'all',
sendRegion: '全国',
connectionStates: [{ status: 'online', currentConnections: 1, desiredConnections: 1 }],
},
},
],
},
});
await expect(service.processSendJob({ messageRecordId: 'record-1' })).resolves.toEqual(
expect.objectContaining({ channelId: 'channel-all' }),
);
expect(gatewayAdd).toHaveBeenCalledWith(
'submit-command',
expect.objectContaining({ channelId: 'channel-all', route: expect.objectContaining({ channelCode: 'CMPP-ALL', carrier: 'mobile' }) }),
);
});
it('records receipts and uplink messages from gateway events', async () => {
const { service, prisma } = createService();
api/src/send-chain/send-chain.service.ts
+74 -15
View File
@@ -464,7 +464,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
try {
const routed = await this.selectChannelForMessage(message);
return this.submitMessageToGateway(message, routed, 0);
return await this.submitMessageToGateway(message, routed, 0);
} catch (error) {
const reason = error instanceof Error ? error.message : '无可用通道组或通道';
await this.prisma.smsMessageRecord.update({
@@ -491,6 +491,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
submittedAt,
},
});
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') {
await this.chargeAcceptedMessage(message);
@@ -538,6 +541,12 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
deliveredAt,
},
});
const isCurrentAttempt =
(!message.channelId || message.channelId === data.channelId)
&& (!message.gatewayMessageId || message.gatewayMessageId === data.gatewayMessageId);
if (!isCurrentAttempt || (status === 'failed' && message.status === 'delivered')) {
return this.prisma.smsMessageRecord.findUnique({ where: { id: message.id } });
}
if (status === 'failed') {
const retried = await this.retryMessageIfAllowed(message, '回执失败补发');
if (retried) {
@@ -619,12 +628,13 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
phoneNumber: string;
content: string;
billingUnits: number;
template?: { signature?: { name?: string | null } | null } | null;
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
},
routed: RoutedChannel,
attempt: number,
) {
const channel = routed.channel;
await this.ensureSignatureReportedForChannel(message, channel.id);
await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
const submitId = `SUB-${randomUUID()}`;
const session = await this.prisma.cmppSubmitSession.upsert({
@@ -733,7 +743,7 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
where: { id: message.id },
data: { errorMessage: reason },
});
return this.submitMessageToGateway(message, routed, attempts.length);
return await this.submitMessageToGateway(message, routed, attempts.length);
} catch {
return null;
}
@@ -750,7 +760,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
const route = await this.findApplicationRoute(message.tenantId, message.applicationId, carrier);
const province = await this.identifyProvince(message.phoneNumber);
const excluded = new Set(options.excludeChannelIds ?? []);
const items = route.group.items.filter((item) => !excluded.has(item.channelId) && isCarrierCompatible(item.channel.carrier, carrier));
const items = route.group.items.filter((item) =>
!excluded.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));
@@ -785,6 +799,9 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (route.group.status !== 'active') {
throw new BadRequestException('企业应用绑定的通道组已停用');
}
if (normalizeCarrier(route.group.carrier) !== carrier) {
throw new BadRequestException('企业应用绑定的通道组运营商与路由规则不一致');
}
return route;
}
@@ -826,17 +843,17 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}
private async resolveUnitPrice(tenantId: string, applicationId?: string) {
try {
const route = await this.prisma.channelRouteRule.findFirst({
where: { tenantId, applicationId, status: 'active', channelId: null },
include: { group: { include: { items: { include: { channel: true }, orderBy: { priority: 'asc' } } } } },
orderBy: { priority: 'asc' },
});
const channel = route?.group.items.find((item) => item.channel.status === 'active')?.channel;
return channel?.unitPrice ?? 0;
} catch {
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 application.customerUnitPrice ?? 0;
}
private async validateSendResources(tenantId: string, applicationId?: string, templateId?: string) {
@@ -935,12 +952,18 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
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: message.amountCents,
smsUnits: message.billingUnits,
relatedType: 'sms_batch_task',
relatedId: message.batchTaskId,
relatedType: 'sms_message_record',
relatedId: message.messageId,
remark: `${remark}: ${message.messageId}`,
});
}
@@ -952,6 +975,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if ((message.amountCents ?? 0) + (message.billingUnits ?? 0) <= 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: message.amountCents,
@@ -966,6 +997,34 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
});
}
private async ensureSignatureReportedForChannel(
message: {
id: string;
templateId?: string | null;
template?: { signature?: { id?: string | null; name?: string | null } | null } | null;
},
channelId: string,
) {
let signatureId = message.template?.signature?.id ?? null;
if (!signatureId && message.templateId) {
const template = await this.prisma.smsTemplate.findUnique({
where: { id: message.templateId },
include: { signature: true },
});
signatureId = template?.signature?.id ?? null;
}
if (!signatureId) {
throw new BadRequestException('短信签名未配置，不能提交到通道');
}
const reportTask = await this.prisma.channelSignatureReportTask.findFirst({
where: { signatureId, channelId, status: 'approved' },
select: { id: true },
});
if (!reportTask) {
throw new BadRequestException('短信签名未在最终通道报备通过');
}
}
private async waitForChannelRateLimit(channelId: string, tps: number) {
const redis = this.getRedis();
for (;;) {