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

fix: close admin channel test sms loop

Browse Source
This commit is contained in:
hectorzhao
2026-07-09 11:49:20 +08:00
parent a4a638208e
commit c58a010951
12 changed files with 582 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/channels/channels.service.ts
+306 -6
View File
@@ -1,5 +1,6 @@
import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Queue } from 'bullmq';
import IORedis from 'ioredis';
import { Prisma } from '@prisma/client';
import { randomUUID } from 'crypto';
import { PrismaService } from '../prisma/prisma.service';
@@ -144,17 +145,30 @@ export interface CopyChannelDto {
operatorId?: string;
}
export interface TestChannelDto {
phoneNumber?: string;
phones?: string[] | string;
content?: string;
accessNo?: string;
operatorId?: string;
}
const GATEWAY_CONNECTION_QUEUE = 'gateway.connection.commands';
const GATEWAY_SUBMIT_QUEUE = 'gateway.submit.queue';
const GATEWAY_SUBMIT_STREAM = 'gateway.submit.commands';
const DEFAULT_GATEWAY_CONTROL_URL = 'http://127.0.0.1:8090';
const DEFAULT_CHANNEL_CONNECTION_ID = 'primary';
const DEFAULT_CONNECTING_TIMEOUT_MS = 30_000;
const DEFAULT_CONNECTING_TIMEOUT_SCAN_MS = 5_000;
const CONNECTING_TIMEOUT_ERROR = 'Gateway connection request timed out';
const DEFAULT_CMPP_VERSION = '2.0';
@Injectable()
export class ChannelsService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(ChannelsService.name);
private gatewayConnectionQueue?: Queue;
private gatewaySubmitQueue?: Queue;
private redis?: IORedis;
private connectionTimeoutTimer?: ReturnType<typeof setInterval>;
constructor(private readonly prisma: PrismaService) {}
@@ -176,6 +190,8 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
clearInterval(this.connectionTimeoutTimer);
}
await this.gatewayConnectionQueue?.close();
await this.gatewaySubmitQueue?.close();
this.redis?.disconnect();
}
listChannels() {
@@ -198,6 +214,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
if (!Number.isInteger(gatewayPort) || gatewayPort <= 0 || gatewayPort > 65535) {
throw new BadRequestException('gatewayPort must be an integer between 1 and 65535');
}
const cmppVersion = normalizeCmppVersion(data.cmppVersion);
const config = normalizeChannelRuntimeConfig(data.config, data.desiredConnections, data.windowSize);
const channel = await this.prisma.smsChannel.create({
data: {
@@ -212,7 +229,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
account: data.account,
passwordCipher: data.passwordCipher,
srcId: data.srcId,
cmppVersion: data.cmppVersion ?? '3.0',
cmppVersion,
rateLimitPerSecond: data.rateLimitPerSecond ?? 100,
unitPrice: data.unitPrice ?? 0,
status: data.status ?? 'active',
@@ -234,6 +251,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
if (gatewayPort !== undefined && (!Number.isInteger(gatewayPort) || gatewayPort <= 0 || gatewayPort > 65535)) {
throw new BadRequestException('gatewayPort must be an integer between 1 and 65535');
}
const cmppVersion = data.cmppVersion === undefined ? undefined : normalizeCmppVersion(data.cmppVersion);
const config = data.config !== undefined || data.desiredConnections !== undefined || data.windowSize !== undefined
? normalizeChannelRuntimeConfig(channel.config, data.desiredConnections, data.windowSize)
: undefined;
@@ -251,7 +269,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
account: data.account,
passwordCipher: data.passwordCipher,
srcId: data.srcId,
cmppVersion: data.cmppVersion,
cmppVersion,
rateLimitPerSecond: data.rateLimitPerSecond,
unitPrice: data.unitPrice,
status: data.status,
@@ -394,11 +412,135 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
return this.changeChannelStatus(channelId, { ...data, status: 'deleted' });
}
testChannel(channelId: string) {
async testChannel(channelId: string, data: TestChannelDto = {}) {
const phoneNumbers = normalizeTestPhones(data);
const content = normalizeTestContent(data.content);
const channel = await this.prisma.smsChannel.findUnique({
where: { id: channelId },
include: { connectionStates: true },
});
if (!channel) {
throw new NotFoundException('Channel not found');
}
if (channel.status !== 'active') {
throw new BadRequestException('通道未启用，不能发送测试短信');
}
const connectedState = channel.connectionStates.find((state) =>
normalizeGatewayConnectionStatus(state.status) === 'connected' && (state.currentConnections ?? 0) > 0,
);
if (!connectedState) {
throw new BadRequestException('通道当前没有可用 CMPP 连接，请先连接成功后再测试发送');
}
const tenant = await this.prisma.tenant.findFirst({ orderBy: { createdAt: 'asc' } });
if (!tenant) {
throw new BadRequestException('未找到可归属测试短信的客户租户');
}
const createdAt = new Date();
const taskNo = `CHTEST-${Date.now()}-${randomUUID().slice(0, 8)}`;
const batchTask = await this.prisma.smsBatchTask.create({
data: {
tenantId: tenant.id,
taskNo,
sourceType: 'admin_channel_test',
content,
category: 'channel_test',
phoneTotal: phoneNumbers.length,
status: 'submit_queued',
auditStatus: 'approved',
progressTotal: phoneNumbers.length,
submittedTotal: 0,
createdById: data.operatorId,
},
});
const results = [];
for (const [index, phoneNumber] of phoneNumbers.entries()) {
const messageId = `MSG-TEST-${Date.now()}-${randomUUID().slice(0, 8)}`;
const submitId = `SUB-TEST-${Date.now()}-${randomUUID().slice(0, 8)}`;
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 },
});
const messageRecord = await this.prisma.smsMessageRecord.create({
data: {
tenantId: tenant.id,
batchTaskId: batchTask.id,
messageId,
phoneNumber,
content,
billingUnits: calculateBillingUnits(content),
unitPrice: channel.unitPrice,
amountCents: channel.unitPrice * calculateBillingUnits(content),
queuePriority: 'normal',
channelId: channel.id,
submitId,
status: 'submit_queued',
submitStatus: 'queued',
errorMessage: '运营端通道测试短信',
},
});
await this.prisma.smsSubmitRecord.create({
data: {
tenantId: tenant.id,
batchTaskId: batchTask.id,
messageRecordId: messageRecord.id,
channelId: channel.id,
sessionId: session.id,
submitId,
submitStatus: 'queued',
},
});
const command = buildChannelTestSubmitCommand({
channel,
content,
phoneNumber,
messageId,
submitId,
batchTaskId: batchTask.id,
tenantId: tenant.id,
attempt: index,
accessNo: data.accessNo,
});
await this.getGatewaySubmitQueue().add('submit-command', command);
const streamMessageId = await this.publishGatewaySubmitCommand(command);
results.push({
phoneNumber,
messageRecordId: messageRecord.id,
submitId,
streamMessageId,
});
}
await this.prisma.smsBatchTask.update({
where: { id: batchTask.id },
data: { submittedTotal: phoneNumbers.length },
});
await this.prisma.operationLog.create({
data: {
userId: data.operatorId,
action: 'sms_channel.test_submit',
resource: 'sms_channel',
resourceId: channel.id,
detail: {
batchTaskId: batchTask.id,
phoneTotal: phoneNumbers.length,
messageRecordIds: results.map((item) => item.messageRecordId),
connectionId: connectedState.connectionId,
} as Prisma.InputJsonValue,
},
});
return {
channelId,
status: 'queued',
message: 'Channel test request accepted as a phase-4 placeholder.',
status: 'submit_queued',
batchTaskId: batchTask.id,
taskNo: batchTask.taskNo,
submitted: results.length,
messages: results,
queuedAt: createdAt,
};
}
@@ -485,7 +627,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
lastDisconnectedAt: data.lastDisconnectedAt ? new Date(data.lastDisconnectedAt) : undefined,
lastHeartbeatAt: data.lastHeartbeatAt ? new Date(data.lastHeartbeatAt) : undefined,
reconnectCount: data.reconnectCount ?? 0,
lastError: data.lastError,
lastError: status === 'connected' ? null : data.lastError,
};
const existing = await this.prisma.cmppConnectionState.findFirst({
where: {
@@ -1038,6 +1180,31 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
return this.gatewayConnectionQueue;
}
private getGatewaySubmitQueue() {
this.gatewaySubmitQueue ??= new Queue(GATEWAY_SUBMIT_QUEUE, { connection: bullmqConnection() });
return this.gatewaySubmitQueue;
}
private getRedis() {
if (!this.redis) {
this.redis = new IORedis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', {
maxRetriesPerRequest: null,
});
}
return this.redis;
}
private async publishGatewaySubmitCommand(command: unknown) {
return this.getRedis().xadd(
process.env.GATEWAY_SUBMIT_STREAM ?? GATEWAY_SUBMIT_STREAM,
'*',
'messageType',
'SubmitCommand',
'data',
JSON.stringify(command),
);
}
private async notifyGatewayConnect(command: Record<string, unknown>) {
const baseUrl = (process.env.GATEWAY_CONTROL_URL ?? DEFAULT_GATEWAY_CONTROL_URL).replace(/\/+$/, '');
let response: { ok: boolean; status: number; text: () => Promise<string> };
@@ -1057,6 +1224,131 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
}
}
function normalizeTestPhones(data: TestChannelDto) {
const rawPhones = Array.isArray(data.phones)
? data.phones
: String(data.phoneNumber ?? data.phones ?? '').split(/[,\n，\s]+/u);
const phones = rawPhones.map((phone) => String(phone).trim()).filter(Boolean);
const uniquePhones = Array.from(new Set(phones));
if (uniquePhones.length === 0) {
throw new BadRequestException('请填写测试手机号');
}
if (uniquePhones.length > 10) {
throw new BadRequestException('测试手机号最多允许 10 个');
}
for (const phone of uniquePhones) {
if (!/^1[3-9]\d{9}$/.test(phone)) {
throw new BadRequestException(`手机号格式不正确：${phone}`);
}
}
return uniquePhones;
}
function normalizeTestContent(content?: string) {
const normalized = (content ?? '').trim();
if (!normalized) {
throw new BadRequestException('请填写测试短信内容');
}
if (normalized.length > 1000) {
throw new BadRequestException('测试短信内容不能超过 1000 字符');
}
return normalized;
}
function calculateBillingUnits(content: string) {
return Math.max(1, Math.ceil([...content].length / 67));
}
function buildChannelTestSubmitCommand({
channel,
content,
phoneNumber,
messageId,
submitId,
batchTaskId,
tenantId,
attempt,
accessNo,
}: {
channel: {
id: string;
code: string;
gatewayHost: string;
gatewayPort: number;
account: string;
passwordCipher: string;
srcId: string;
cmppVersion: string;
rateLimitPerSecond: number;
config?: Prisma.JsonValue | null;
};
content: string;
phoneNumber: string;
messageId: string;
submitId: string;
batchTaskId: string;
tenantId: string;
attempt: number;
accessNo?: string;
}) {
const srcId = accessNo?.trim() ? `${channel.srcId}${accessNo.trim()}` : channel.srcId;
return {
schemaVersion: 'v1',
messageType: 'SubmitCommand',
traceId: randomUUID(),
messageId,
channelId: channel.id,
createdAt: new Date().toISOString(),
tenantId,
applicationId: 'admin-channel-test',
taskId: batchTaskId,
submitId,
queuePriority: 'normal',
phoneNumber,
content,
signature: 'CHANNEL_TEST',
templateId: 'admin-channel-test',
billingUnits: calculateBillingUnits(content),
route: {
channelCode: channel.code,
cmppAccountCode: channel.account,
priority: attempt,
rateLimitPerSecond: channel.rateLimitPerSecond,
},
cmpp: {
serviceId: getStringConfigValue(channel.config, 'serviceId', 'SMS'),
srcId,
registeredDelivery: 1,
msgFmt: 8,
},
upstream: {
gatewayHost: channel.gatewayHost,
gatewayPort: channel.gatewayPort,
account: channel.account,
passwordCipher: channel.passwordCipher,
cmppVersion: channel.cmppVersion,
desiredConnections: getPositiveRuntimeInteger(getConfigValue(channel.config, 'desiredConnections'), 1, 'desiredConnections'),
windowSize: getPositiveRuntimeInteger(getConfigValue(channel.config, 'windowSize'), 16, 'windowSize'),
},
retry: { attempt: 0, maxAttempts: 1 },
};
}
function getConfigValue(config: Prisma.JsonValue | null | undefined, key: string) {
if (config && typeof config === 'object' && !Array.isArray(config) && key in config) {
return config[key as keyof typeof config];
}
return undefined;
}
function getStringConfigValue(config: Prisma.JsonValue | null | undefined, key: string, fallback: string) {
const value = getConfigValue(config, key);
if (value === undefined || value === null || value === '') {
return fallback;
}
return String(value);
}
function normalizeConnectionAction(status: string) {
const normalized = status.toLowerCase();
if (normalized === 'connected') {
@@ -1077,6 +1369,14 @@ function normalizeConnectionAction(status: string) {
return 'updated';
}
function normalizeCmppVersion(version?: string) {
const normalized = (version ?? DEFAULT_CMPP_VERSION).trim();
if (normalized === '2.0' || normalized === '3.0') {
return normalized;
}
throw new BadRequestException('cmppVersion must be 2.0 or 3.0');
}
function normalizeGatewayConnectionStatus(status: string) {
const normalized = status.toLowerCase();
if (['online', 'open', 'connected'].includes(normalized)) {