1674 lines
55 KiB
TypeScript
1674 lines
55 KiB
TypeScript
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';
|
||
|
||
export interface CreateChannelDto {
|
||
code: string;
|
||
name: string;
|
||
carrier?: string;
|
||
sendRegion?: string;
|
||
protocol?: string;
|
||
gatewayHost: string;
|
||
gatewayPort: number;
|
||
enterpriseCode?: string;
|
||
account: string;
|
||
passwordCipher: string;
|
||
srcId: string;
|
||
cmppVersion?: string;
|
||
rateLimitPerSecond?: number;
|
||
unitPrice?: number;
|
||
status?: string;
|
||
desiredConnections?: number;
|
||
windowSize?: number;
|
||
config?: Record<string, unknown>;
|
||
}
|
||
|
||
export type UpdateChannelDto = Partial<CreateChannelDto>;
|
||
|
||
export interface CreateChannelGroupDto {
|
||
code: string;
|
||
name: string;
|
||
carrier: string;
|
||
description?: string;
|
||
status?: string;
|
||
retryEnabled?: boolean;
|
||
retryTimeLimitHours?: number;
|
||
retryTimeLimitMinutes?: number;
|
||
}
|
||
|
||
export interface CreateChannelGroupItemDto {
|
||
groupId: string;
|
||
channelId: string;
|
||
carrier?: string;
|
||
province?: string;
|
||
priority?: number;
|
||
weight?: number;
|
||
isBackup?: boolean;
|
||
rateLimitPerSecond?: number;
|
||
}
|
||
|
||
export interface UpdateChannelGroupDto {
|
||
code?: string;
|
||
name?: string;
|
||
carrier?: string;
|
||
description?: string;
|
||
status?: string;
|
||
retryEnabled?: boolean;
|
||
retryTimeLimitHours?: number;
|
||
retryTimeLimitMinutes?: number;
|
||
items?: Array<Omit<CreateChannelGroupItemDto, 'groupId'>>;
|
||
}
|
||
|
||
export interface CreateRouteRuleDto {
|
||
tenantId?: string;
|
||
applicationId?: string;
|
||
groupId: string;
|
||
channelId?: string;
|
||
carrier?: string;
|
||
province?: string;
|
||
priority?: number;
|
||
status?: string;
|
||
}
|
||
|
||
export interface CreateReportFieldDto {
|
||
channelId: string;
|
||
drainageFieldId: string;
|
||
reportType: 'signature' | 'drainage' | 'both';
|
||
code?: string;
|
||
name?: string;
|
||
fieldType?: string;
|
||
required?: boolean;
|
||
description?: string;
|
||
sortOrder?: number;
|
||
status?: string;
|
||
}
|
||
|
||
export interface CreateReportMaterialDto {
|
||
signatureId: string;
|
||
channelId: string;
|
||
fieldCode: string;
|
||
fieldValue?: string;
|
||
fileObjectId?: string;
|
||
}
|
||
|
||
export interface CreateReportTaskDto {
|
||
tenantId: string;
|
||
signatureId: string;
|
||
channelId: string;
|
||
createdById?: string;
|
||
}
|
||
|
||
export interface CreateReportExportDto {
|
||
fileObjectId?: string;
|
||
fileName: string;
|
||
rowCount?: number;
|
||
}
|
||
|
||
export interface CreateReceiptImportDto {
|
||
fileObjectId?: string;
|
||
fileName: string;
|
||
fileContent?: string;
|
||
delimiter?: ',' | '\t';
|
||
rowCount?: number;
|
||
successCount?: number;
|
||
failedCount?: number;
|
||
statusAfter?: string;
|
||
reason?: string;
|
||
result?: Record<string, unknown>;
|
||
}
|
||
|
||
export interface UpsertConnectionStateDto {
|
||
tenantId?: string;
|
||
applicationId?: string;
|
||
channelId: string;
|
||
connectionId: string;
|
||
status: string;
|
||
desiredConnections?: number;
|
||
currentConnections?: number;
|
||
lastConnectedAt?: string;
|
||
lastDisconnectedAt?: string;
|
||
lastHeartbeatAt?: string;
|
||
reconnectCount?: number;
|
||
lastError?: string;
|
||
}
|
||
|
||
export interface ChangeChannelStatusDto {
|
||
status: string;
|
||
operatorId?: string;
|
||
reason?: string;
|
||
}
|
||
|
||
export interface CopyChannelDto {
|
||
name?: string;
|
||
code?: string;
|
||
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) {}
|
||
|
||
onModuleInit() {
|
||
if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED === 'true') {
|
||
return;
|
||
}
|
||
this.connectionTimeoutTimer = setInterval(() => {
|
||
void this.markTimedOutConnectingChannels().catch((error) => {
|
||
this.logger.error(`Failed to mark timed-out CMPP connections: ${error instanceof Error ? error.message : String(error)}`);
|
||
});
|
||
}, getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_SCAN_MS', DEFAULT_CONNECTING_TIMEOUT_SCAN_MS));
|
||
this.connectionTimeoutTimer.unref?.();
|
||
}
|
||
|
||
async onModuleDestroy() {
|
||
if (this.connectionTimeoutTimer) {
|
||
clearInterval(this.connectionTimeoutTimer);
|
||
}
|
||
await this.gatewayConnectionQueue?.close();
|
||
await this.gatewaySubmitQueue?.close();
|
||
this.redis?.disconnect();
|
||
}
|
||
|
||
listChannels() {
|
||
return this.prisma.smsChannel.findMany({
|
||
include: { connectionStates: true },
|
||
orderBy: { createdAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
async createChannel(data: CreateChannelDto) {
|
||
const missingFields = ['code', 'name', 'gatewayHost', 'gatewayPort', 'account', 'passwordCipher', 'srcId'].filter((field) => {
|
||
const value = data[field as keyof CreateChannelDto];
|
||
return value === undefined || value === null || value === '';
|
||
});
|
||
if (missingFields.length > 0) {
|
||
throw new BadRequestException(`Missing required channel fields: ${missingFields.join(', ')}`);
|
||
}
|
||
const gatewayPort = Number(data.gatewayPort);
|
||
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(undefined, data.config, data.desiredConnections, data.windowSize);
|
||
const rateLimitPerSecond = normalizeChannelRateLimit(data.rateLimitPerSecond);
|
||
const channel = await this.prisma.smsChannel.create({
|
||
data: {
|
||
code: data.code,
|
||
name: data.name,
|
||
carrier: data.carrier,
|
||
sendRegion: data.sendRegion ?? '全国',
|
||
protocol: data.protocol ?? 'CMPP',
|
||
gatewayHost: data.gatewayHost,
|
||
gatewayPort,
|
||
enterpriseCode: data.enterpriseCode,
|
||
account: data.account,
|
||
passwordCipher: data.passwordCipher,
|
||
srcId: data.srcId,
|
||
cmppVersion,
|
||
rateLimitPerSecond,
|
||
unitPrice: data.unitPrice ?? 0,
|
||
status: data.status ?? 'active',
|
||
config: config as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
if (channel.status === 'active') {
|
||
await this.requestChannelConnection(channel, 'channel_created');
|
||
}
|
||
return channel;
|
||
}
|
||
|
||
async updateChannel(channelId: string, data: UpdateChannelDto) {
|
||
const channel = await this.prisma.smsChannel.findUnique({ where: { id: channelId } });
|
||
if (!channel) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
const gatewayPort = data.gatewayPort === undefined ? undefined : Number(data.gatewayPort);
|
||
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.config, data.desiredConnections, data.windowSize)
|
||
: undefined;
|
||
const rateLimitPerSecond = data.rateLimitPerSecond === undefined
|
||
? undefined
|
||
: normalizeChannelRateLimit(data.rateLimitPerSecond);
|
||
const updated = await this.prisma.smsChannel.update({
|
||
where: { id: channelId },
|
||
data: {
|
||
code: data.code,
|
||
name: data.name,
|
||
carrier: data.carrier,
|
||
sendRegion: data.sendRegion,
|
||
protocol: data.protocol,
|
||
gatewayHost: data.gatewayHost,
|
||
gatewayPort,
|
||
enterpriseCode: data.enterpriseCode,
|
||
account: data.account,
|
||
passwordCipher: data.passwordCipher,
|
||
srcId: data.srcId,
|
||
cmppVersion,
|
||
rateLimitPerSecond,
|
||
unitPrice: data.unitPrice,
|
||
status: data.status,
|
||
config: config as Prisma.InputJsonValue | undefined,
|
||
},
|
||
});
|
||
await this.prisma.operationLog.create({
|
||
data: {
|
||
action: 'sms_channel.update',
|
||
resource: 'sms_channel',
|
||
resourceId: channelId,
|
||
detail: {
|
||
before: {
|
||
code: channel.code,
|
||
name: channel.name,
|
||
carrier: channel.carrier,
|
||
sendRegion: channel.sendRegion,
|
||
gatewayHost: channel.gatewayHost,
|
||
gatewayPort: channel.gatewayPort,
|
||
enterpriseCode: channel.enterpriseCode,
|
||
account: channel.account,
|
||
srcId: channel.srcId,
|
||
unitPrice: channel.unitPrice,
|
||
},
|
||
after: data,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
return updated;
|
||
}
|
||
|
||
async changeChannelStatus(channelId: string, data: ChangeChannelStatusDto) {
|
||
const channel = await this.prisma.smsChannel.findUnique({ where: { id: channelId } });
|
||
if (!channel) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
const updated = await this.prisma.smsChannel.update({ where: { id: channelId }, data: { status: data.status } });
|
||
await this.prisma.operationLog.create({
|
||
data: {
|
||
userId: data.operatorId,
|
||
action: `sms_channel.${data.status}`,
|
||
resource: 'sms_channel',
|
||
resourceId: channelId,
|
||
detail: {
|
||
statusBefore: channel.status,
|
||
statusAfter: data.status,
|
||
reason: data.reason,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
if (data.status === 'active') {
|
||
await this.requestChannelConnection(updated, 'channel_enabled', data.operatorId);
|
||
}
|
||
return updated;
|
||
}
|
||
|
||
async copyChannel(channelId: string, data: CopyChannelDto = {}) {
|
||
const source = await this.prisma.smsChannel.findUnique({
|
||
where: { id: channelId },
|
||
include: { reportFields: true },
|
||
});
|
||
if (!source) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
|
||
const suffix = Date.now().toString(36).toUpperCase();
|
||
const nextName = data.name ?? `${source.name}副本`;
|
||
const nextCode = data.code ?? `${source.code}-COPY-${suffix}`;
|
||
|
||
const copied = await this.prisma.$transaction(async (tx) => {
|
||
const nextChannel = await tx.smsChannel.create({
|
||
data: {
|
||
code: nextCode,
|
||
name: nextName,
|
||
carrier: source.carrier,
|
||
protocol: source.protocol,
|
||
gatewayHost: source.gatewayHost,
|
||
gatewayPort: source.gatewayPort,
|
||
enterpriseCode: source.enterpriseCode,
|
||
account: source.account,
|
||
passwordCipher: source.passwordCipher,
|
||
srcId: source.srcId,
|
||
sendRegion: source.sendRegion,
|
||
cmppVersion: source.cmppVersion,
|
||
rateLimitPerSecond: source.rateLimitPerSecond,
|
||
unitPrice: source.unitPrice,
|
||
status: 'disabled',
|
||
config: source.config as Prisma.InputJsonValue | undefined,
|
||
reportFields: {
|
||
create: source.reportFields.map((field) => ({
|
||
drainageFieldId: field.drainageFieldId,
|
||
reportType: field.reportType,
|
||
code: field.code,
|
||
name: field.name,
|
||
fieldType: field.fieldType,
|
||
required: field.required,
|
||
description: field.description,
|
||
sortOrder: field.sortOrder,
|
||
status: field.status,
|
||
})),
|
||
},
|
||
},
|
||
include: { reportFields: true },
|
||
});
|
||
|
||
const reportMaterials = await tx.signatureReportMaterial.findMany({ where: { channelId } });
|
||
if (reportMaterials.length > 0) {
|
||
await tx.signatureReportMaterial.createMany({
|
||
data: reportMaterials.map((material) => ({
|
||
signatureId: material.signatureId,
|
||
channelId: nextChannel.id,
|
||
fieldCode: material.fieldCode,
|
||
fieldValue: material.fieldValue,
|
||
fileObjectId: material.fileObjectId,
|
||
})),
|
||
skipDuplicates: true,
|
||
});
|
||
}
|
||
|
||
await tx.operationLog.create({
|
||
data: {
|
||
userId: data.operatorId,
|
||
action: 'sms_channel.copy',
|
||
resource: 'sms_channel',
|
||
resourceId: nextChannel.id,
|
||
detail: {
|
||
sourceChannelId: source.id,
|
||
sourceCode: source.code,
|
||
sourceStatus: source.status,
|
||
copiedStatus: 'disabled',
|
||
copiedReportFields: source.reportFields.length,
|
||
copiedReportMaterials: reportMaterials.length,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
|
||
return nextChannel;
|
||
});
|
||
|
||
return copied;
|
||
}
|
||
|
||
async deleteChannel(channelId: string, data: ChangeChannelStatusDto = { status: 'deleted' }) {
|
||
return this.changeChannelStatus(channelId, { ...data, status: 'deleted' });
|
||
}
|
||
|
||
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 createdAt = new Date();
|
||
const testNo = `CHTEST-${Date.now()}-${randomUUID().slice(0, 8)}`;
|
||
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: {
|
||
messageId,
|
||
phoneNumber,
|
||
content,
|
||
billingUnits: calculateBillingUnits(content),
|
||
unitPrice: 0,
|
||
amountCents: 0,
|
||
queuePriority: 'normal',
|
||
channelId: channel.id,
|
||
submitId,
|
||
status: 'submit_queued',
|
||
submitStatus: 'queued',
|
||
errorMessage: '运营端通道测试短信',
|
||
},
|
||
});
|
||
await this.prisma.smsSubmitRecord.create({
|
||
data: {
|
||
messageRecordId: messageRecord.id,
|
||
channelId: channel.id,
|
||
sessionId: session.id,
|
||
submitId,
|
||
submitStatus: 'queued',
|
||
},
|
||
});
|
||
const command = buildChannelTestSubmitCommand({
|
||
channel,
|
||
content,
|
||
phoneNumber,
|
||
messageId,
|
||
submitId,
|
||
testNo,
|
||
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.operationLog.create({
|
||
data: {
|
||
userId: data.operatorId,
|
||
action: 'sms_channel.test_submit',
|
||
resource: 'sms_channel',
|
||
resourceId: channel.id,
|
||
detail: {
|
||
testNo,
|
||
phoneTotal: phoneNumbers.length,
|
||
messageRecordIds: results.map((item) => item.messageRecordId),
|
||
connectionId: connectedState.connectionId,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
|
||
return {
|
||
channelId,
|
||
status: 'submit_queued',
|
||
testNo,
|
||
submitted: results.length,
|
||
messages: results,
|
||
queuedAt: createdAt,
|
||
};
|
||
}
|
||
|
||
listChannelMetrics(channelId: string) {
|
||
return this.prisma.channelHealthMetric.findMany({
|
||
where: { channelId },
|
||
orderBy: { windowStart: 'desc' },
|
||
take: 100,
|
||
});
|
||
}
|
||
|
||
listChannelConnections(channelId: string) {
|
||
return this.prisma.cmppConnectionState.findMany({
|
||
where: { channelId },
|
||
orderBy: { updatedAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
async listChannelConnectionLogs(channelId: string) {
|
||
const channel = await this.prisma.smsChannel.findUnique({ where: { id: channelId }, select: { id: true } });
|
||
if (!channel) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
const [connectionStates, logs] = await Promise.all([
|
||
this.prisma.cmppConnectionState.findMany({
|
||
where: { channelId },
|
||
orderBy: { updatedAt: 'desc' },
|
||
take: 50,
|
||
}),
|
||
this.prisma.operationLog.findMany({
|
||
where: {
|
||
OR: [
|
||
{ resource: 'sms_channel', resourceId: channelId },
|
||
{ resource: 'cmpp_connection', resourceId: { startsWith: `${channelId}:` } },
|
||
],
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
take: 100,
|
||
}),
|
||
]);
|
||
return {
|
||
channelId,
|
||
connectionStates,
|
||
logs: logs.map((log) => ({
|
||
id: log.id,
|
||
time: log.createdAt,
|
||
event: normalizeLinkEvent(log.action),
|
||
action: log.action,
|
||
resourceId: log.resourceId,
|
||
detail: log.detail,
|
||
})),
|
||
};
|
||
}
|
||
|
||
listTenantConnections(tenantId: string) {
|
||
return this.prisma.cmppConnectionState.findMany({
|
||
where: { tenantId },
|
||
include: { channel: true },
|
||
orderBy: { updatedAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
async upsertConnectionState(data: UpsertConnectionStateDto) {
|
||
const status = normalizeGatewayConnectionStatus(data.status);
|
||
if (data.applicationId) {
|
||
const application = await this.prisma.smsApplication.findUnique({ where: { id: data.applicationId }, select: { tenantId: true } });
|
||
if (!application) {
|
||
throw new BadRequestException('applicationId does not reference an existing application');
|
||
}
|
||
if (data.tenantId && data.tenantId !== application.tenantId) {
|
||
throw new BadRequestException('applicationId does not belong to tenantId');
|
||
}
|
||
data.tenantId = application.tenantId;
|
||
}
|
||
const payload = {
|
||
tenantId: data.tenantId,
|
||
applicationId: data.applicationId,
|
||
status,
|
||
desiredConnections: data.desiredConnections ?? 1,
|
||
currentConnections: data.currentConnections ?? (status === 'connected' ? 1 : 0),
|
||
lastConnectedAt: data.lastConnectedAt ? new Date(data.lastConnectedAt) : undefined,
|
||
lastDisconnectedAt: data.lastDisconnectedAt ? new Date(data.lastDisconnectedAt) : undefined,
|
||
lastHeartbeatAt: data.lastHeartbeatAt ? new Date(data.lastHeartbeatAt) : undefined,
|
||
reconnectCount: data.reconnectCount ?? 0,
|
||
lastError: status === 'connected' ? null : data.lastError,
|
||
};
|
||
const existing = await this.prisma.cmppConnectionState.findFirst({
|
||
where: {
|
||
applicationId: data.applicationId ?? null,
|
||
channelId: data.channelId,
|
||
connectionId: data.connectionId,
|
||
},
|
||
});
|
||
const state = existing
|
||
? await this.prisma.cmppConnectionState.update({ where: { id: existing.id }, data: payload })
|
||
: await this.prisma.cmppConnectionState.create({
|
||
data: {
|
||
channelId: data.channelId,
|
||
connectionId: data.connectionId,
|
||
...payload,
|
||
},
|
||
});
|
||
await this.prisma.operationLog.create({
|
||
data: {
|
||
tenantId: data.tenantId,
|
||
action: `cmpp_connection.${normalizeConnectionAction(status)}`,
|
||
resource: 'cmpp_connection',
|
||
resourceId: `${data.channelId}:${data.connectionId}`,
|
||
detail: {
|
||
status,
|
||
applicationId: state.applicationId,
|
||
desiredConnections: state.desiredConnections,
|
||
currentConnections: state.currentConnections,
|
||
lastError: state.lastError,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
return state;
|
||
}
|
||
|
||
async markTimedOutConnectingChannels(now = new Date()) {
|
||
const timeoutMs = getPositiveIntegerEnv('GATEWAY_CONNECTING_TIMEOUT_MS', DEFAULT_CONNECTING_TIMEOUT_MS);
|
||
const cutoff = new Date(now.getTime() - timeoutMs);
|
||
const lastError = `${CONNECTING_TIMEOUT_ERROR} after ${Math.round(timeoutMs / 1000)} seconds`;
|
||
const states = await this.prisma.cmppConnectionState.findMany({
|
||
where: {
|
||
status: 'connecting',
|
||
updatedAt: { lte: cutoff },
|
||
},
|
||
select: {
|
||
id: true,
|
||
tenantId: true,
|
||
applicationId: true,
|
||
channelId: true,
|
||
connectionId: true,
|
||
desiredConnections: true,
|
||
currentConnections: true,
|
||
updatedAt: true,
|
||
},
|
||
take: 100,
|
||
});
|
||
let failed = 0;
|
||
for (const state of states) {
|
||
const result = await this.prisma.cmppConnectionState.updateMany({
|
||
where: {
|
||
id: state.id,
|
||
status: 'connecting',
|
||
updatedAt: { lte: cutoff },
|
||
},
|
||
data: {
|
||
status: 'failed',
|
||
currentConnections: 0,
|
||
lastDisconnectedAt: now,
|
||
lastError,
|
||
},
|
||
});
|
||
if (result.count === 0) {
|
||
continue;
|
||
}
|
||
failed += result.count;
|
||
await this.prisma.operationLog.create({
|
||
data: {
|
||
tenantId: state.tenantId,
|
||
action: 'cmpp_connection.failed',
|
||
resource: 'cmpp_connection',
|
||
resourceId: `${state.channelId}:${state.connectionId}`,
|
||
detail: {
|
||
reason: 'connect_timeout',
|
||
applicationId: state.applicationId,
|
||
status: 'failed',
|
||
previousStatus: 'connecting',
|
||
desiredConnections: state.desiredConnections,
|
||
currentConnectionsBefore: state.currentConnections,
|
||
currentConnections: 0,
|
||
timeoutMs,
|
||
lastError,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
}
|
||
return { checked: states.length, failed };
|
||
}
|
||
|
||
listGroups() {
|
||
return this.prisma.smsChannelGroup.findMany({
|
||
include: { items: { include: { channel: { include: { connectionStates: true } } }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
|
||
orderBy: { createdAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
createGroup(data: CreateChannelGroupDto) {
|
||
const retryTimeLimitMinutes = normalizeRetryTimeLimitMinutes(data.retryTimeLimitMinutes, data.retryTimeLimitHours, 720);
|
||
const carrier = normalizeBusinessCarrier(data.carrier);
|
||
return this.prisma.smsChannelGroup.create({
|
||
data: {
|
||
code: data.code,
|
||
name: data.name,
|
||
carrier,
|
||
description: data.description,
|
||
status: data.status ?? 'active',
|
||
retryEnabled: data.retryEnabled ?? true,
|
||
retryTimeLimitHours: Math.ceil(retryTimeLimitMinutes / 60),
|
||
retryTimeLimitMinutes,
|
||
},
|
||
});
|
||
}
|
||
|
||
async addGroupItem(data: CreateChannelGroupItemDto) {
|
||
const group = await this.prisma.smsChannelGroup.findUnique({ where: { id: data.groupId } });
|
||
if (!group) {
|
||
throw new NotFoundException('Channel group not found');
|
||
}
|
||
const groupCarrier = normalizeBusinessCarrier(group.carrier);
|
||
const itemCarrier = data.carrier ? normalizeBusinessCarrier(data.carrier) : groupCarrier;
|
||
if (itemCarrier !== groupCarrier) {
|
||
throw new BadRequestException('Channel group items must use the same carrier as the channel group');
|
||
}
|
||
const channel = await this.prisma.smsChannel.findUnique({ where: { id: data.channelId } });
|
||
if (!channel) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
if (!isChannelCarrierCompatible(channel.carrier, groupCarrier)) {
|
||
throw new BadRequestException('Channel carrier is not compatible with the channel group carrier');
|
||
}
|
||
if (data.province && !isRegionCompatible(channel.sendRegion, data.province)) {
|
||
throw new BadRequestException('Province route must use a channel with the same sendRegion');
|
||
}
|
||
const existing = await this.prisma.smsChannelGroupItem.findFirst({
|
||
where: { groupId: data.groupId, channelId: data.channelId },
|
||
});
|
||
if (existing) {
|
||
throw new BadRequestException('通道组内不能重复配置同一通道');
|
||
}
|
||
if (data.province) {
|
||
const existingProvince = await this.prisma.smsChannelGroupItem.findFirst({
|
||
where: { groupId: data.groupId, province: data.province },
|
||
});
|
||
if (existingProvince) {
|
||
throw new BadRequestException('同一通道组内同一省份只能配置一个通道');
|
||
}
|
||
} else {
|
||
const existingPriority = await this.prisma.smsChannelGroupItem.findFirst({
|
||
where: { groupId: data.groupId, province: null, priority: data.priority ?? 100 },
|
||
});
|
||
if (existingPriority) {
|
||
throw new BadRequestException('同一通道组内全国通道优先级不能重复');
|
||
}
|
||
}
|
||
return this.prisma.smsChannelGroupItem.create({
|
||
data: {
|
||
groupId: data.groupId,
|
||
channelId: data.channelId,
|
||
carrier: itemCarrier,
|
||
province: data.province,
|
||
priority: data.priority ?? 100,
|
||
weight: data.weight ?? 1,
|
||
isBackup: data.isBackup ?? false,
|
||
rateLimitPerSecond: data.rateLimitPerSecond,
|
||
},
|
||
});
|
||
}
|
||
|
||
async updateGroup(groupId: string, data: UpdateChannelGroupDto) {
|
||
const current = await this.prisma.smsChannelGroup.findUnique({ where: { id: groupId } });
|
||
if (!current) {
|
||
throw new NotFoundException('Channel group not found');
|
||
}
|
||
const retryTimeLimitMinutes = normalizeRetryTimeLimitMinutes(
|
||
data.retryTimeLimitMinutes,
|
||
data.retryTimeLimitHours,
|
||
current.retryTimeLimitMinutes ?? current.retryTimeLimitHours * 60,
|
||
);
|
||
const carrier = data.carrier ? normalizeBusinessCarrier(data.carrier) : normalizeBusinessCarrier(current.carrier);
|
||
const items = data.items ?? [];
|
||
const channelIds = [...new Set(items.map((item) => item.channelId))];
|
||
const channels = await this.prisma.smsChannel.findMany({ where: { id: { in: channelIds } } });
|
||
const channelById = new Map(channels.map((channel) => [channel.id, channel]));
|
||
validateGroupItems(carrier, items, channelById);
|
||
|
||
return this.prisma.$transaction(async (tx) => {
|
||
await tx.smsChannelGroupItem.deleteMany({ where: { groupId } });
|
||
await tx.smsChannelGroup.update({
|
||
where: { id: groupId },
|
||
data: {
|
||
code: data.code ?? current.code,
|
||
name: data.name ?? current.name,
|
||
carrier,
|
||
description: data.description,
|
||
status: data.status ?? current.status,
|
||
retryEnabled: data.retryEnabled ?? current.retryEnabled,
|
||
retryTimeLimitHours: Math.ceil(retryTimeLimitMinutes / 60),
|
||
retryTimeLimitMinutes,
|
||
},
|
||
});
|
||
if (items.length > 0) {
|
||
await tx.smsChannelGroupItem.createMany({
|
||
data: items.map((item) => ({
|
||
groupId,
|
||
channelId: item.channelId,
|
||
carrier,
|
||
province: item.province,
|
||
priority: item.priority ?? 100,
|
||
weight: item.weight ?? 1,
|
||
isBackup: item.isBackup ?? false,
|
||
rateLimitPerSecond: item.rateLimitPerSecond,
|
||
})),
|
||
});
|
||
}
|
||
return tx.smsChannelGroup.findUnique({
|
||
where: { id: groupId },
|
||
include: { items: { include: { channel: true }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
|
||
});
|
||
});
|
||
}
|
||
|
||
async deleteGroup(groupId: string) {
|
||
const group = await this.prisma.smsChannelGroup.findUnique({ where: { id: groupId } });
|
||
if (!group) {
|
||
throw new NotFoundException('Channel group not found');
|
||
}
|
||
const boundRoute = await this.prisma.channelRouteRule.findFirst({
|
||
where: {
|
||
groupId,
|
||
status: 'active',
|
||
},
|
||
select: { id: true },
|
||
});
|
||
if (boundRoute) {
|
||
throw new BadRequestException('Channel group is used by application route rules and cannot be deleted');
|
||
}
|
||
await this.prisma.smsChannelGroupItem.deleteMany({ where: { groupId } });
|
||
return this.prisma.smsChannelGroup.delete({ where: { id: groupId } });
|
||
}
|
||
|
||
listRouteRules() {
|
||
return this.prisma.channelRouteRule.findMany({
|
||
include: { group: true, channel: true },
|
||
orderBy: [{ priority: 'asc' }, { createdAt: 'desc' }],
|
||
});
|
||
}
|
||
|
||
async createRouteRule(data: CreateRouteRuleDto) {
|
||
if (!data.applicationId) {
|
||
throw new BadRequestException('applicationId is required for channel group routing');
|
||
}
|
||
if (!data.carrier) {
|
||
throw new BadRequestException('carrier is required for application channel group routing');
|
||
}
|
||
const carrier = normalizeBusinessCarrier(data.carrier);
|
||
if (data.channelId) {
|
||
throw new BadRequestException('Route rules can only bind channel groups, not single channels');
|
||
}
|
||
if (data.province) {
|
||
throw new BadRequestException('Province routing must be configured inside the channel group');
|
||
}
|
||
const group = await this.prisma.smsChannelGroup.findUnique({ where: { id: data.groupId } });
|
||
if (!group) {
|
||
throw new NotFoundException('Channel group not found');
|
||
}
|
||
if (normalizeBusinessCarrier(group.carrier) !== carrier) {
|
||
throw new BadRequestException('Route rule carrier must match the channel group carrier');
|
||
}
|
||
return this.prisma.channelRouteRule.create({
|
||
data: {
|
||
tenantId: data.tenantId,
|
||
applicationId: data.applicationId,
|
||
groupId: data.groupId,
|
||
channelId: undefined,
|
||
carrier,
|
||
province: undefined,
|
||
priority: data.priority ?? 100,
|
||
status: data.status ?? 'active',
|
||
},
|
||
});
|
||
}
|
||
|
||
listReportFields(channelId?: string) {
|
||
return this.prisma.channelReportField.findMany({
|
||
where: channelId ? { channelId } : undefined,
|
||
include: { drainageField: true },
|
||
orderBy: [{ sortOrder: 'asc' }, { createdAt: 'desc' }],
|
||
});
|
||
}
|
||
|
||
async createReportField(data: CreateReportFieldDto) {
|
||
const field = await this.prisma.drainageField.findUnique({ where: { id: data.drainageFieldId } });
|
||
if (!field || field.status !== 'active') {
|
||
throw new BadRequestException('报备字段库字段不存在或已停用');
|
||
}
|
||
const reportType = normalizeReportType(data.reportType);
|
||
return this.prisma.channelReportField.create({
|
||
data: {
|
||
channelId: data.channelId,
|
||
drainageFieldId: field.id,
|
||
reportType,
|
||
code: field.code,
|
||
name: field.name,
|
||
fieldType: field.fieldType,
|
||
required: data.required ?? field.required,
|
||
description: data.description ?? field.description,
|
||
sortOrder: data.sortOrder ?? 100,
|
||
status: data.status ?? 'active',
|
||
},
|
||
});
|
||
}
|
||
|
||
listReportMaterials(signatureId?: string, channelId?: string) {
|
||
return this.prisma.signatureReportMaterial.findMany({
|
||
where: {
|
||
signatureId,
|
||
channelId,
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
upsertReportMaterial(data: CreateReportMaterialDto) {
|
||
return this.prisma.signatureReportMaterial.upsert({
|
||
where: {
|
||
signatureId_channelId_fieldCode: {
|
||
signatureId: data.signatureId,
|
||
channelId: data.channelId,
|
||
fieldCode: data.fieldCode,
|
||
},
|
||
},
|
||
update: {
|
||
fieldValue: data.fieldValue,
|
||
fileObjectId: data.fileObjectId,
|
||
},
|
||
create: {
|
||
signatureId: data.signatureId,
|
||
channelId: data.channelId,
|
||
fieldCode: data.fieldCode,
|
||
fieldValue: data.fieldValue,
|
||
fileObjectId: data.fileObjectId,
|
||
},
|
||
});
|
||
}
|
||
|
||
listReportTasks(tenantId?: string, status?: string) {
|
||
return this.prisma.channelSignatureReportTask.findMany({
|
||
where: { tenantId, status },
|
||
include: { signature: true, channel: true },
|
||
orderBy: { createdAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
async createReportTask(data: CreateReportTaskDto) {
|
||
const task = await this.prisma.channelSignatureReportTask.create({
|
||
data: {
|
||
tenantId: data.tenantId,
|
||
signatureId: data.signatureId,
|
||
channelId: data.channelId,
|
||
createdById: data.createdById,
|
||
status: 'pending',
|
||
},
|
||
});
|
||
await this.recordReportTask(task.id, task.channelId, 'create', undefined, 'pending');
|
||
return task;
|
||
}
|
||
|
||
async createReportExport(taskId: string, data: CreateReportExportDto) {
|
||
const task = await this.getReportTaskOrThrow(taskId);
|
||
const exported = await this.prisma.reportExportFile.create({
|
||
data: {
|
||
taskId,
|
||
fileObjectId: data.fileObjectId,
|
||
fileName: data.fileName,
|
||
rowCount: data.rowCount ?? 0,
|
||
},
|
||
});
|
||
await this.updateReportTaskStatus(taskId, task.channelId, task.status, 'exporting', 'export');
|
||
return exported;
|
||
}
|
||
|
||
async importReportReceipt(taskId: string, data: CreateReceiptImportDto) {
|
||
const task = await this.getReportTaskOrThrow(taskId);
|
||
const parsed = data.fileContent ? parseReceiptContent(data.fileContent, data.delimiter) : undefined;
|
||
const rowCount = data.rowCount ?? parsed?.rowCount ?? 0;
|
||
const successCount = data.successCount ?? parsed?.successCount ?? 0;
|
||
const failedCount = data.failedCount ?? parsed?.failedCount ?? 0;
|
||
const statusAfter = data.statusAfter ?? deriveReceiptStatus(rowCount, successCount, failedCount);
|
||
const imported = await this.prisma.reportReceiptImport.create({
|
||
data: {
|
||
taskId,
|
||
fileObjectId: data.fileObjectId,
|
||
fileName: data.fileName,
|
||
rowCount,
|
||
successCount,
|
||
failedCount,
|
||
status: 'imported',
|
||
result: (data.result ?? parsed?.result) as Prisma.InputJsonValue | undefined,
|
||
},
|
||
});
|
||
await this.updateReportTaskStatus(taskId, task.channelId, task.status, statusAfter, 'receipt_import', data.reason);
|
||
await this.prisma.smsSignature.update({
|
||
where: { id: task.signatureId },
|
||
data: { reportStatus: statusAfter },
|
||
});
|
||
return imported;
|
||
}
|
||
|
||
listReportRecords(taskId?: string, channelId?: string) {
|
||
return this.prisma.channelSignatureReportRecord.findMany({
|
||
where: { taskId, channelId },
|
||
orderBy: { createdAt: 'desc' },
|
||
});
|
||
}
|
||
|
||
private async getReportTaskOrThrow(taskId: string) {
|
||
const task = await this.prisma.channelSignatureReportTask.findUnique({ where: { id: taskId } });
|
||
if (!task) {
|
||
throw new NotFoundException('Report task not found');
|
||
}
|
||
return task;
|
||
}
|
||
|
||
private async updateReportTaskStatus(
|
||
taskId: string,
|
||
channelId: string,
|
||
statusBefore: string,
|
||
statusAfter: string,
|
||
action: string,
|
||
reason?: string,
|
||
) {
|
||
await this.prisma.channelSignatureReportTask.update({
|
||
where: { id: taskId },
|
||
data: { status: statusAfter, reason },
|
||
});
|
||
await this.recordReportTask(taskId, channelId, action, statusBefore, statusAfter, reason);
|
||
}
|
||
|
||
private recordReportTask(
|
||
taskId: string,
|
||
channelId: string,
|
||
action: string,
|
||
statusBefore: string | undefined,
|
||
statusAfter: string,
|
||
reason?: string,
|
||
) {
|
||
return this.prisma.channelSignatureReportRecord.create({
|
||
data: {
|
||
taskId,
|
||
channelId,
|
||
action,
|
||
statusBefore,
|
||
statusAfter,
|
||
reason,
|
||
},
|
||
});
|
||
}
|
||
|
||
private async requestChannelConnection(
|
||
channel: {
|
||
id: string;
|
||
code: string;
|
||
name: string;
|
||
gatewayHost: string;
|
||
gatewayPort: number;
|
||
account: string;
|
||
passwordCipher: string;
|
||
srcId: string;
|
||
cmppVersion: string;
|
||
rateLimitPerSecond: number;
|
||
config?: Prisma.JsonValue | null;
|
||
},
|
||
reason: 'channel_created' | 'channel_enabled',
|
||
operatorId?: string,
|
||
) {
|
||
const desiredConnections = getDesiredConnections(channel.config);
|
||
const connectionId = defaultChannelConnectionId(channel.id);
|
||
const existing = await this.prisma.cmppConnectionState.findFirst({
|
||
where: {
|
||
applicationId: null,
|
||
channelId: channel.id,
|
||
connectionId,
|
||
},
|
||
});
|
||
const data = {
|
||
applicationId: null,
|
||
status: 'connecting',
|
||
desiredConnections,
|
||
currentConnections: 0,
|
||
lastError: null,
|
||
};
|
||
const state = existing
|
||
? await this.prisma.cmppConnectionState.update({ where: { id: existing.id }, data })
|
||
: await this.prisma.cmppConnectionState.create({
|
||
data: {
|
||
channelId: channel.id,
|
||
connectionId,
|
||
...data,
|
||
},
|
||
});
|
||
await this.prisma.operationLog.create({
|
||
data: {
|
||
userId: operatorId,
|
||
action: 'cmpp_connection.connect_requested',
|
||
resource: 'cmpp_connection',
|
||
resourceId: `${channel.id}:${connectionId}`,
|
||
detail: {
|
||
reason,
|
||
status: state.status,
|
||
desiredConnections: state.desiredConnections,
|
||
currentConnections: state.currentConnections,
|
||
} as Prisma.InputJsonValue,
|
||
},
|
||
});
|
||
const command = {
|
||
schemaVersion: 'v1',
|
||
messageType: 'ConnectChannel',
|
||
traceId: randomUUID(),
|
||
channelId: channel.id,
|
||
connectionId,
|
||
createdAt: new Date().toISOString(),
|
||
reason,
|
||
desiredConnections,
|
||
channel: {
|
||
code: channel.code,
|
||
name: channel.name,
|
||
gatewayHost: channel.gatewayHost,
|
||
gatewayPort: channel.gatewayPort,
|
||
account: channel.account,
|
||
passwordCipher: channel.passwordCipher,
|
||
srcId: channel.srcId,
|
||
cmppVersion: channel.cmppVersion,
|
||
rateLimitPerSecond: channel.rateLimitPerSecond,
|
||
},
|
||
};
|
||
await this.getGatewayConnectionQueue().add('connect-channel', command, { jobId: `${connectionId}:connect` });
|
||
await this.notifyGatewayConnect(command);
|
||
return state;
|
||
}
|
||
|
||
private getGatewayConnectionQueue() {
|
||
this.gatewayConnectionQueue ??= new Queue(GATEWAY_CONNECTION_QUEUE, { connection: bullmqConnection() });
|
||
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> };
|
||
try {
|
||
response = await fetch(`${baseUrl}/connections/connect`, {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify(command),
|
||
});
|
||
} catch (error) {
|
||
throw new BadRequestException(`Gateway connect request failed: ${error instanceof Error ? error.message : String(error)}`);
|
||
}
|
||
if (!response.ok) {
|
||
const responseText = await response.text();
|
||
throw new BadRequestException(`Gateway connect request failed: ${response.status} ${responseText}`);
|
||
}
|
||
}
|
||
}
|
||
|
||
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,
|
||
testNo,
|
||
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;
|
||
testNo: 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: 'platform-channel-test',
|
||
applicationId: 'admin-channel-test',
|
||
taskId: testNo,
|
||
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,
|
||
extensionDigits: normalizeExtensionDigits(getConfigValue(channel.config, 'extensionDigits')),
|
||
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') {
|
||
return 'connected';
|
||
}
|
||
if (['heartbeat', 'active_test'].includes(normalized)) {
|
||
return 'heartbeat';
|
||
}
|
||
if (['reconnecting', 'reconnect'].includes(normalized)) {
|
||
return 'reconnecting';
|
||
}
|
||
if (['offline', 'closed', 'disconnected'].includes(normalized)) {
|
||
return 'disconnected';
|
||
}
|
||
if (['auth_failed', 'heartbeat_timeout', 'failed', 'error'].includes(normalized)) {
|
||
return 'failed';
|
||
}
|
||
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)) {
|
||
return 'connected';
|
||
}
|
||
if (['connecting', 'connect_requested'].includes(normalized)) {
|
||
return 'connecting';
|
||
}
|
||
if (['reconnecting', 'reconnect'].includes(normalized)) {
|
||
return 'reconnecting';
|
||
}
|
||
if (['offline', 'closed', 'disconnected'].includes(normalized)) {
|
||
return 'disconnected';
|
||
}
|
||
if (['auth_failed', 'heartbeat_timeout', 'failed', 'error'].includes(normalized)) {
|
||
return 'failed';
|
||
}
|
||
return normalized;
|
||
}
|
||
|
||
function defaultChannelConnectionId(channelId: string) {
|
||
return `${channelId}:${DEFAULT_CHANNEL_CONNECTION_ID}`;
|
||
}
|
||
|
||
function getDesiredConnections(config?: Prisma.JsonValue | null) {
|
||
if (config && typeof config === 'object' && !Array.isArray(config) && 'desiredConnections' in config) {
|
||
const value = Number(config.desiredConnections);
|
||
if (Number.isInteger(value) && value > 0) {
|
||
return value;
|
||
}
|
||
}
|
||
return 1;
|
||
}
|
||
|
||
function normalizeChannelRuntimeConfig(
|
||
existingConfig?: Prisma.JsonValue | Record<string, unknown> | null,
|
||
incomingConfig?: Record<string, unknown> | null,
|
||
desiredConnections?: number,
|
||
windowSize?: number,
|
||
) {
|
||
const existing = existingConfig && typeof existingConfig === 'object' && !Array.isArray(existingConfig)
|
||
? existingConfig as Record<string, unknown>
|
||
: {};
|
||
const incoming = incomingConfig && typeof incomingConfig === 'object' && !Array.isArray(incomingConfig)
|
||
? incomingConfig
|
||
: {};
|
||
const base = { ...existing, ...incoming };
|
||
base.desiredConnections = getPositiveRuntimeInteger(desiredConnections ?? base.desiredConnections, 1, 'desiredConnections');
|
||
base.windowSize = getPositiveRuntimeInteger(windowSize ?? base.windowSize, 16, 'windowSize');
|
||
base.extensionDigits = normalizeExtensionDigits(base.extensionDigits);
|
||
return base;
|
||
}
|
||
|
||
function normalizeChannelRateLimit(value: unknown) {
|
||
const normalized = getPositiveRuntimeInteger(value, 100, 'rateLimitPerSecond');
|
||
if (normalized > 2000) {
|
||
throw new BadRequestException('rateLimitPerSecond must be between 1 and 2000');
|
||
}
|
||
return normalized;
|
||
}
|
||
|
||
function normalizeExtensionDigits(value: unknown) {
|
||
if (value === undefined || value === null || value === '') {
|
||
return 0;
|
||
}
|
||
const normalized = Number(value);
|
||
if (!Number.isInteger(normalized) || normalized < 0 || normalized > 20) {
|
||
throw new BadRequestException('extensionDigits must be an integer between 0 and 20');
|
||
}
|
||
return normalized;
|
||
}
|
||
|
||
function getPositiveRuntimeInteger(value: unknown, fallback: number, fieldName: string) {
|
||
if (value === undefined || value === null || value === '') {
|
||
return fallback;
|
||
}
|
||
const normalized = Number(value);
|
||
if (!Number.isInteger(normalized) || normalized <= 0) {
|
||
throw new BadRequestException(`${fieldName} must be a positive integer`);
|
||
}
|
||
return normalized;
|
||
}
|
||
|
||
function bullmqConnection() {
|
||
const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379');
|
||
return {
|
||
host: redisUrl.hostname,
|
||
port: Number(redisUrl.port || 6379),
|
||
username: redisUrl.username || undefined,
|
||
password: redisUrl.password || undefined,
|
||
maxRetriesPerRequest: null,
|
||
};
|
||
}
|
||
|
||
function getPositiveIntegerEnv(name: string, fallback: number) {
|
||
const value = Number(process.env[name]);
|
||
if (Number.isInteger(value) && value > 0) {
|
||
return value;
|
||
}
|
||
return fallback;
|
||
}
|
||
|
||
|
||
function parseReceiptContent(content: string, delimiter?: ',' | '\t') {
|
||
const lines = content.replace(/^\uFEFF/, '').split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
|
||
if (lines.length === 0) {
|
||
throw new BadRequestException('Receipt file is empty');
|
||
}
|
||
const separator = delimiter ?? (lines[0].includes('\t') ? '\t' : ',');
|
||
const firstCells = splitReceiptLine(lines[0], separator);
|
||
const hasHeader = firstCells.some((cell) => ['phone', 'mobile', 'status', 'result', '手机号', '号码', '状态', '结果'].includes(cell.toLowerCase()));
|
||
const header = hasHeader ? firstCells : [];
|
||
const rows = hasHeader ? lines.slice(1) : lines;
|
||
const statusIndex = findReceiptStatusIndex(header);
|
||
let successCount = 0;
|
||
let failedCount = 0;
|
||
const resultRows = rows.map((line, index) => {
|
||
const cells = splitReceiptLine(line, separator);
|
||
const rawStatus = cells[statusIndex] ?? cells[cells.length - 1] ?? '';
|
||
const normalizedStatus = normalizeReceiptStatus(rawStatus);
|
||
if (normalizedStatus === 'success') {
|
||
successCount += 1;
|
||
} else {
|
||
failedCount += 1;
|
||
}
|
||
return {
|
||
rowNumber: (hasHeader ? index + 2 : index + 1),
|
||
phone: cells[0] ?? '',
|
||
status: normalizedStatus,
|
||
rawStatus,
|
||
raw: cells,
|
||
};
|
||
});
|
||
return {
|
||
rowCount: resultRows.length,
|
||
successCount,
|
||
failedCount,
|
||
result: {
|
||
delimiter: separator === '\t' ? 'tab' : 'comma',
|
||
hasHeader,
|
||
rows: resultRows,
|
||
},
|
||
};
|
||
}
|
||
|
||
function splitReceiptLine(line: string, delimiter: ',' | '\t') {
|
||
if (delimiter === '\t') {
|
||
return line.split('\t').map((cell) => stripReceiptCell(cell));
|
||
}
|
||
const cells: string[] = [];
|
||
let current = '';
|
||
let quoted = false;
|
||
for (let index = 0; index < line.length; index += 1) {
|
||
const char = line[index];
|
||
const next = line[index + 1];
|
||
if (char === '"' && quoted && next === '"') {
|
||
current += '"';
|
||
index += 1;
|
||
} else if (char === '"') {
|
||
quoted = !quoted;
|
||
} else if (char === ',' && !quoted) {
|
||
cells.push(stripReceiptCell(current));
|
||
current = '';
|
||
} else {
|
||
current += char;
|
||
}
|
||
}
|
||
cells.push(stripReceiptCell(current));
|
||
return cells;
|
||
}
|
||
|
||
function stripReceiptCell(value: string) {
|
||
return value.trim().replace(/^"|"$/g, '').trim();
|
||
}
|
||
|
||
function findReceiptStatusIndex(header: string[]) {
|
||
if (header.length === 0) {
|
||
return 1;
|
||
}
|
||
const index = header.findIndex((cell) => ['status', 'result', '状态', '结果'].includes(cell.toLowerCase()));
|
||
return index >= 0 ? index : Math.max(0, header.length - 1);
|
||
}
|
||
|
||
function normalizeReceiptStatus(value: string) {
|
||
const normalized = value.trim().toLowerCase();
|
||
if (['success', 'succeeded', 'approved', 'completed', 'ok', 'pass', 'passed', '通过', '成功', '已完成', '报备成功'].includes(normalized)) {
|
||
return 'success';
|
||
}
|
||
if (['failed', 'fail', 'rejected', 'reject', 'error', 'no', 'denied', '驳回', '失败', '不通过', '拒绝', '报备失败'].includes(normalized)) {
|
||
return 'failed';
|
||
}
|
||
return 'failed';
|
||
}
|
||
|
||
function deriveReceiptStatus(rowCount: number, successCount: number, failedCount: number) {
|
||
if (rowCount <= 0 || successCount <= 0) {
|
||
return 'failed';
|
||
}
|
||
if (failedCount > 0) {
|
||
return 'partial';
|
||
}
|
||
return 'completed';
|
||
}
|
||
|
||
function normalizeRetryTimeLimitMinutes(minutes: number | undefined, hours: number | undefined, fallbackMinutes: number) {
|
||
const value = minutes ?? (hours === undefined ? fallbackMinutes : hours * 60);
|
||
if (!Number.isInteger(value) || value <= 0 || value > 72 * 60) {
|
||
throw new BadRequestException('retryTimeLimitMinutes must be an integer between 1 and 4320');
|
||
}
|
||
return value;
|
||
}
|
||
|
||
function normalizeBusinessCarrier(carrier?: string | null) {
|
||
const normalized = normalizeChannelCarrier(carrier);
|
||
if (!['mobile', 'unicom', 'telecom'].includes(normalized)) {
|
||
throw new BadRequestException('carrier must be mobile, unicom, or telecom');
|
||
}
|
||
return normalized;
|
||
}
|
||
|
||
function normalizeChannelCarrier(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;
|
||
}
|
||
|
||
function isChannelCarrierCompatible(channelCarrier: string | null | undefined, groupCarrier: string) {
|
||
const normalized = normalizeChannelCarrier(channelCarrier);
|
||
return normalized === 'all' || normalized === groupCarrier;
|
||
}
|
||
|
||
function normalizeRegion(region?: string | null) {
|
||
return String(region ?? '').replace(/省|市|自治区|壮族|回族|维吾尔/g, '').trim();
|
||
}
|
||
|
||
function isRegionCompatible(channelRegion: string | null | undefined, itemProvince: string) {
|
||
return normalizeRegion(channelRegion) === normalizeRegion(itemProvince);
|
||
}
|
||
|
||
function validateGroupItems(
|
||
groupCarrier: string,
|
||
items: Array<Omit<CreateChannelGroupItemDto, 'groupId'>>,
|
||
channels: Map<string, { id: string; carrier?: string | null; sendRegion?: string | null }>,
|
||
) {
|
||
const channelIds = new Set<string>();
|
||
const provinces = new Set<string>();
|
||
const nationalPriorities = new Set<number>();
|
||
for (const item of items) {
|
||
const itemCarrier = item.carrier ? normalizeBusinessCarrier(item.carrier) : groupCarrier;
|
||
if (itemCarrier !== groupCarrier) {
|
||
throw new BadRequestException('Channel group items must use the same carrier as the channel group');
|
||
}
|
||
const channel = channels.get(item.channelId);
|
||
if (!channel) {
|
||
throw new NotFoundException('Channel not found');
|
||
}
|
||
if (channelIds.has(item.channelId)) {
|
||
throw new BadRequestException('通道组内不能重复配置同一通道');
|
||
}
|
||
channelIds.add(item.channelId);
|
||
if (!isChannelCarrierCompatible(channel.carrier, groupCarrier)) {
|
||
throw new BadRequestException('Channel carrier is not compatible with the channel group carrier');
|
||
}
|
||
if (item.province) {
|
||
const province = normalizeRegion(item.province);
|
||
if (provinces.has(province)) {
|
||
throw new BadRequestException('同一通道组内同一省份只能配置一个通道');
|
||
}
|
||
provinces.add(province);
|
||
if (!isRegionCompatible(channel.sendRegion, item.province)) {
|
||
throw new BadRequestException('Province route must use a channel with the same sendRegion');
|
||
}
|
||
} else {
|
||
const priority = item.priority ?? 100;
|
||
if (nationalPriorities.has(priority)) {
|
||
throw new BadRequestException('同一通道组内全国通道优先级不能重复');
|
||
}
|
||
nationalPriorities.add(priority);
|
||
}
|
||
}
|
||
}
|
||
|
||
function normalizeReportType(value?: string) {
|
||
if (value === 'signature' || value === 'drainage' || value === 'both') return value;
|
||
throw new BadRequestException('reportType must be signature, drainage or both');
|
||
}
|
||
|
||
function normalizeLinkEvent(action: string) {
|
||
if (action.includes('connect_requested')) {
|
||
return '连接请求';
|
||
}
|
||
if (action.includes('connected')) {
|
||
return '连接成功';
|
||
}
|
||
if (action.includes('heartbeat')) {
|
||
return '心跳';
|
||
}
|
||
if (action.includes('reconnecting')) {
|
||
return '重连';
|
||
}
|
||
if (action.includes('disconnected')) {
|
||
return '断开';
|
||
}
|
||
if (action.includes('failed')) {
|
||
return '连接失败';
|
||
}
|
||
if (action.includes('copy')) {
|
||
return '复制';
|
||
}
|
||
if (action.includes('deleted')) {
|
||
return '删除';
|
||
}
|
||
return '更新';
|
||
}
|