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

fix: implement channel group routing failover

Browse Source
This commit is contained in:
hectorzhao
2026-07-03 11:07:43 +08:00
parent 131f344ac4
commit a890b03163
16 changed files with 853 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.ts
+301 -73
View File
@@ -80,6 +80,25 @@ interface SendJob {
messageRecordId: string;
}
type RoutedChannel = {
channel: {
id: string;
code: string;
account: string;
srcId: string;
rateLimitPerSecond: number;
unitPrice: number;
status: string;
carrier?: string | null;
sendRegion: string;
config?: unknown;
};
carrier: string;
province?: string | null;
groupId: string;
routeScope: 'province' | 'national';
};
const SEND_QUEUE = 'sms.send.queue';
const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue';
@@ -443,63 +462,19 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (!message || message.status !== 'queued') {
return { skipped: true };
}
const channel = await this.selectChannel(message.tenantId, message.applicationId ?? undefined);
await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
const submitId = `SUB-${randomUUID()}`;
const session = await this.prisma.cmppSubmitSession.upsert({
where: { sessionNo: `OPEN-${channel.id}` },
update: { submitTotal: { increment: 1 } },
create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
});
await this.prisma.smsSubmitRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
channelId: channel.id,
sessionId: session.id,
submitId,
submitStatus: 'queued',
},
});
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { channelId: channel.id, submitId, status: 'submit_queued', submitStatus: 'queued' },
});
await this.getGatewayQueue().add('submit-command', {
schemaVersion: 'v1',
messageType: 'SubmitCommand',
traceId: randomUUID(),
messageId: message.messageId,
channelId: channel.id,
createdAt: new Date().toISOString(),
tenantId: message.tenantId,
applicationId: message.applicationId ?? 'unknown',
taskId: message.batchTaskId,
submitId,
phoneNumber: message.phoneNumber,
content: message.content,
signature: message.template?.signature?.name ?? 'SMS',
templateId: message.templateId ?? 'unknown',
billingUnits: message.billingUnits,
route: {
channelCode: channel.code,
cmppAccountCode: channel.account,
priority: 0,
rateLimitPerSecond: channel.rateLimitPerSecond,
},
cmpp: {
serviceId: channel.config && typeof channel.config === 'object' && 'serviceId' in channel.config
? String(channel.config.serviceId)
: 'SMS',
srcId: channel.srcId,
registeredDelivery: 1,
msgFmt: 8,
},
retry: { attempt: 0, maxAttempts: 3 },
});
await this.refreshTaskProgress(message.batchTaskId);
return { submitted: true, messageRecordId: message.id, channelId: channel.id };
try {
const routed = await this.selectChannelForMessage(message);
return this.submitMessageToGateway(message, routed, 0);
} catch (error) {
const reason = error instanceof Error ? error.message : '无可用通道组或通道';
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { status: 'failed', errorMessage: reason },
});
await this.releaseMessageReservation(message, reason);
await this.refreshTaskProgress(message.batchTaskId);
return { submitted: false, messageRecordId: message.id, status: 'failed', reason };
}
}
async handleSubmitResult(data: GatewaySubmitResultDto) {
@@ -520,6 +495,11 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if (data.submitStatus === 'accepted') {
await this.chargeAcceptedMessage(message);
} else {
const retried = await this.retryMessageIfAllowed(message, data.submitStatus === 'timeout' ? '提交超时补发' : '提交失败补发');
if (retried) {
await this.refreshTaskProgress(message.batchTaskId);
return retried;
}
await this.releaseMessageReservation(message, data.submitStatus === 'timeout' ? '提交超时释放冻结' : '提交失败释放冻结');
}
await this.prisma.smsMessageRecord.update({
@@ -543,9 +523,6 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
const deliveredAt = data.deliveredAt ? new Date(data.deliveredAt) : new Date();
const status =
data.receiptStatus === 'delivered' ? 'delivered' : data.receiptStatus === 'unknown' ? 'unknown' : 'failed';
if (status === 'failed') {
await this.refundMessage(message, '最终失败退款');
}
await this.prisma.smsReceiptRecord.create({
data: {
tenantId: message.tenantId,
@@ -561,6 +538,14 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
deliveredAt,
},
});
if (status === 'failed') {
const retried = await this.retryMessageIfAllowed(message, '回执失败补发');
if (retried) {
await this.refreshTaskProgress(message.batchTaskId);
return retried;
}
await this.refundMessage(message, '最终失败退款');
}
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: {
@@ -623,30 +608,232 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
return { timeout: candidates.length };
}
private async selectChannel(tenantId: string, applicationId?: string) {
private async submitMessageToGateway(
message: {
id: string;
tenantId: string;
batchTaskId: string;
applicationId?: string | null;
templateId?: string | null;
messageId: string;
phoneNumber: string;
content: string;
billingUnits: number;
template?: { signature?: { name?: string | null } | null } | null;
},
routed: RoutedChannel,
attempt: number,
) {
const channel = routed.channel;
await this.waitForChannelRateLimit(channel.id, channel.rateLimitPerSecond);
const submitId = `SUB-${randomUUID()}`;
const session = await this.prisma.cmppSubmitSession.upsert({
where: { sessionNo: `OPEN-${channel.id}` },
update: { submitTotal: { increment: 1 } },
create: { channelId: channel.id, sessionNo: `OPEN-${channel.id}`, submitTotal: 1 },
});
await this.prisma.smsSubmitRecord.create({
data: {
tenantId: message.tenantId,
batchTaskId: message.batchTaskId,
messageRecordId: message.id,
channelId: channel.id,
sessionId: session.id,
submitId,
submitStatus: 'queued',
},
});
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: {
channelId: channel.id,
submitId,
status: 'submit_queued',
submitStatus: 'queued',
receiptStatus: null,
errorCode: null,
errorMessage: attempt > 0 ? `第 ${attempt + 1} 次提交，路由至${routed.routeScope === 'national' ? '全国' : '省网'}通道` : undefined,
},
});
await this.getGatewayQueue().add('submit-command', {
schemaVersion: 'v1',
messageType: 'SubmitCommand',
traceId: randomUUID(),
messageId: message.messageId,
channelId: channel.id,
createdAt: new Date().toISOString(),
tenantId: message.tenantId,
applicationId: message.applicationId ?? 'unknown',
taskId: message.batchTaskId,
submitId,
phoneNumber: message.phoneNumber,
content: message.content,
signature: message.template?.signature?.name ?? 'SMS',
templateId: message.templateId ?? 'unknown',
billingUnits: message.billingUnits,
route: {
channelCode: channel.code,
cmppAccountCode: channel.account,
priority: attempt,
rateLimitPerSecond: channel.rateLimitPerSecond,
carrier: routed.carrier,
province: routed.province ?? undefined,
scope: routed.routeScope,
groupId: routed.groupId,
},
cmpp: {
serviceId: channel.config && typeof channel.config === 'object' && 'serviceId' in channel.config
? String(channel.config.serviceId)
: 'SMS',
srcId: channel.srcId,
registeredDelivery: 1,
msgFmt: 8,
},
retry: { attempt, maxAttempts: 0 },
});
await this.refreshTaskProgress(message.batchTaskId);
return { submitted: true, messageRecordId: message.id, channelId: channel.id, attempt };
}
private async retryMessageIfAllowed(
message: {
id: string;
tenantId: string;
batchTaskId: string;
applicationId?: string | null;
templateId?: string | null;
messageId: string;
phoneNumber: string;
content: string;
billingUnits: number;
queuedAt?: Date;
},
reason: string,
) {
const attempts = await this.prisma.smsSubmitRecord.findMany({
where: { messageRecordId: message.id },
orderBy: { createdAt: 'asc' },
take: 200,
});
const attemptedChannelIds = attempts.map((attempt) => attempt.channelId);
const ageHours = (Date.now() - new Date(message.queuedAt ?? Date.now()).getTime()) / 3_600_000;
if (ageHours >= 72) {
return null;
}
const route = await this.findApplicationRoute(message.tenantId, message.applicationId ?? undefined, await this.identifyCarrier(message.phoneNumber));
if (!route.group.retryEnabled || ageHours >= Math.min(route.group.retryTimeLimitHours, 72)) {
return null;
}
try {
const routed = await this.selectChannelForMessage(message, {
forceNational: true,
excludeChannelIds: attemptedChannelIds,
});
await this.prisma.smsMessageRecord.update({
where: { id: message.id },
data: { errorMessage: reason },
});
return this.submitMessageToGateway(message, routed, attempts.length);
} catch {
return null;
}
}
private async selectChannelForMessage(
message: { tenantId: string; applicationId?: string | null; phoneNumber: string },
options: { forceNational?: boolean; excludeChannelIds?: string[] } = {},
): Promise<RoutedChannel> {
if (!message.applicationId) {
throw new BadRequestException('短信应用未配置，无法选择通道组');
}
const carrier = await this.identifyCarrier(message.phoneNumber);
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 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));
if (!selected) {
throw new NotFoundException('无可用在线通道');
}
return {
channel: selected.channel,
carrier,
province,
groupId: route.groupId,
routeScope: isNationalChannel(selected) ? 'national' : 'province',
};
}
private async findApplicationRoute(tenantId: string, applicationId: string | undefined, carrier: string) {
const route = await this.prisma.channelRouteRule.findFirst({
where: {
status: 'active',
OR: [{ tenantId, applicationId }, { tenantId, applicationId: null }, { tenantId: null, applicationId: null }],
tenantId,
applicationId,
carrier,
channelId: null,
province: null,
},
include: { channel: true, group: { include: { items: { include: { channel: true }, orderBy: { priority: 'asc' } } } } },
include: { group: { include: { items: { include: { channel: { include: { connectionStates: true } } }, orderBy: { priority: 'asc' } } } } },
orderBy: { priority: 'asc' },
});
const routedChannel = route?.channel ?? route?.group.items.find((item) => item.channel.status === 'active')?.channel;
if (routedChannel) {
return routedChannel;
if (!route) {
throw new NotFoundException('企业应用未配置对应运营商通道组');
}
const channel = await this.prisma.smsChannel.findFirst({ where: { status: 'active' }, orderBy: { createdAt: 'asc' } });
if (!channel) {
throw new NotFoundException('No active SMS channel available');
if (route.group.status !== 'active') {
throw new BadRequestException('企业应用绑定的通道组已停用');
}
return channel;
return route;
}
private async identifyCarrier(phoneNumber: string) {
const rules = await this.prisma.phoneCarrierRule.findMany({
where: { status: 'active' },
orderBy: [{ priority: 'asc' }, { createdAt: 'asc' }],
take: 100,
});
for (const rule of rules) {
try {
if (new RegExp(rule.pattern).test(phoneNumber)) {
return normalizeCarrier(rule.carrier);
}
} catch {
continue;
}
}
return 'mobile';
}
private async identifyProvince(phoneNumber: string) {
for (let length = Math.min(7, phoneNumber.length); length >= 3; length -= 1) {
const segment = await this.prisma.phoneSegment.findUnique({ where: { prefix: phoneNumber.slice(0, length) } });
if (segment?.province) {
return segment.province;
}
}
return null;
}
private isChannelSendAvailable(channel: { status: string; connectionStates?: Array<{ status: string; currentConnections: number; desiredConnections: number }> }) {
if (channel.status !== 'active') {
return false;
}
return (channel.connectionStates ?? []).some((connection) =>
connection.desiredConnections > 0 && connection.currentConnections > 0 && ['online', 'connected'].includes(connection.status),
);
}
private async resolveUnitPrice(tenantId: string, applicationId?: string) {
try {
const channel = await this.selectChannel(tenantId, applicationId);
return channel.unitPrice ?? 0;
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 {
return 0;
}
@@ -695,6 +882,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
}) {
const amountCents = message.amountCents ?? 0;
const smsUnits = message.billingUnits ?? 0;
const exists = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId } });
if (exists?.billingStatus === 'charged') {
return;
}
if (amountCents + smsUnits > 0) {
await this.billing.release({
tenantId: message.tenantId,
@@ -713,7 +904,6 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
relatedId: message.messageId,
remark: '提交成功扣费',
});
const exists = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId } });
const data = {
tenantId: message.tenantId,
applicationId: message.applicationId ?? undefined,
@@ -741,6 +931,10 @@ export class SendChainService implements OnModuleInit, OnModuleDestroy {
if ((message.amountCents ?? 0) + (message.billingUnits ?? 0) <= 0) {
return;
}
const charged = await this.prisma.smsBillingRecord.findFirst({ where: { messageId: message.messageId, billingStatus: 'charged' } });
if (charged) {
return;
}
await this.billing.release({
tenantId: message.tenantId,
amountCents: message.amountCents,
@@ -914,6 +1108,40 @@ function cellByHeader(headers: string[], cells: string[], candidates: string[])
return index >= 0 ? cells[index] : undefined;
}
function normalizeCarrier(carrier?: string | null) {
const value = String(carrier ?? '').trim().toLowerCase();
if (['mobile', 'cmcc', '移动', '中国移动'].includes(value)) return 'mobile';
if (['unicom', 'cucc', '联通', '中国联通'].includes(value)) return 'unicom';
if (['telecom', 'ctcc', '电信', '中国电信'].includes(value)) return 'telecom';
if (['all', 'tri', '三网', '全网'].includes(value)) return 'all';
return value || 'mobile';
}
function isCarrierCompatible(channelCarrier: string | null | undefined, targetCarrier: string) {
const normalized = normalizeCarrier(channelCarrier);
return normalized === 'all' || normalized === targetCarrier;
}
function normalizeRegion(region?: string | null) {
return String(region ?? '').replace(/省|市|自治区|壮族|回族|维吾尔/g, '').trim();
}
function isNationalChannel(item: { province?: string | null; channel: { sendRegion?: string | null } }) {
const itemProvince = normalizeRegion(item.province);
const sendRegion = normalizeRegion(item.channel.sendRegion);
return !itemProvince || itemProvince === '全国' || !sendRegion || sendRegion === '全国';
}
function isProvinceChannel(item: { province?: string | null; channel: { sendRegion?: string | null } }, province?: string | null) {
if (!province) {
return false;
}
const target = normalizeRegion(province);
const itemProvince = normalizeRegion(item.province);
const sendRegion = normalizeRegion(item.channel.sendRegion);
return itemProvince === target || sendRegion === target;
}
function bullmqConnection() {
const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
return {