251 lines
13 KiB
TypeScript
251 lines
13 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 { assertMoneyUnits, moneyToNumber } from '../common/money';
|
|
import { PrismaService } from '../prisma/prisma.service';
|
|
import type { CreateChannelDto, UpdateChannelDto, CreateChannelGroupDto, CreateChannelGroupItemDto, UpdateChannelGroupDto, CreateRouteRuleDto, CreateReportFieldDto, ReplaceReportFieldsDto, CreateReportMaterialDto, CreateReportTaskDto, ChangeReportTaskStatusesDto, CreateReportExportDto, CreateReceiptImportDto, UpsertConnectionStateDto, ChangeChannelStatusDto, CopyChannelDto, TestChannelDto } from './channels.contracts';
|
|
import { GATEWAY_CONNECTION_QUEUE, GATEWAY_SUBMIT_QUEUE, GATEWAY_SUBMIT_STREAM, DEFAULT_GATEWAY_CONTROL_URL, DEFAULT_CHANNEL_CONNECTION_ID, DEFAULT_CONNECTING_TIMEOUT_MS, DEFAULT_CONNECTING_TIMEOUT_SCAN_MS, DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS, DEFAULT_GATEWAY_RECONCILE_INTERVAL_MS, DEFAULT_GATEWAY_CONTROL_TIMEOUT_MS, DEFAULT_HEARTBEAT_INTERVAL_SECONDS, DEFAULT_HEARTBEAT_MISS_THRESHOLD, HEARTBEAT_AUDIT_INTERVAL_MS, CONNECTING_TIMEOUT_ERROR, DEFAULT_CMPP_VERSION, normalizeTestPhones, normalizeTestContent, calculateBillingUnits, buildChannelTestSubmitCommand, getConfigValue, getStringConfigValue, normalizeConnectionAction, normalizeCmppVersion, normalizeGatewayConnectionStatus, defaultChannelConnectionId, getDesiredConnections, ChannelConnectionSettings, getRuntimeConfigInteger, channelConnectionSettingsChanged, channelGroupAuditSnapshot, normalizeChannelRuntimeConfig, normalizeCmppServiceId, normalizeChannelRateLimit, normalizeExtensionDigits, getPositiveRuntimeInteger, bullmqConnection, getPositiveIntegerEnv, parseReceiptContent, splitReceiptLine, stripReceiptCell, findReceiptStatusIndex, normalizeReceiptStatus, deriveReceiptStatus, ChannelReportDeliveryRow, summarizeChannelReportDelivery, sumReportDelivery, percentage, latestDate, currentShanghaiDayRange, normalizeRetryTimeLimitMinutes, normalizeSpreadsheetSize, normalizeBusinessCarrier, normalizeChannelCarrier, normalizeChannelCarriers, legacyCarrierFromCapabilities, isChannelCarrierCompatible, normalizeRegion, isRegionCompatible, validateGroupItems, normalizeReportType, summarizeReportStatuses, normalizeLinkEvent } from './channels.helpers';
|
|
import { ChannelConnectionService } from './channel-connection.service';
|
|
|
|
/** R5 channel domain service composed behind ChannelsService. */
|
|
export class ChannelConfigurationService {
|
|
constructor(private readonly prisma: PrismaService, private readonly connection: ChannelConnectionService) {}
|
|
|
|
listChannels() {
|
|
return this.prisma.smsChannel.findMany({
|
|
include: { connectionStates: true },
|
|
orderBy: { createdAt: 'desc' },
|
|
});
|
|
}
|
|
|
|
async listChannelsPage(query: { keyword?: string; carrier?: string; status?: string; page?: number; pageSize?: number }) {
|
|
const page = Math.max(1, Math.floor(Number(query.page) || 1));
|
|
const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 10)));
|
|
const where: Prisma.SmsChannelWhereInput = {
|
|
status: query.status && query.status !== 'all' ? query.status : { not: 'deleted' },
|
|
carriers: query.carrier && query.carrier !== 'all' ? { has: normalizeBusinessCarrier(query.carrier) } : undefined,
|
|
name: query.keyword?.trim() ? { contains: query.keyword.trim() } : undefined,
|
|
};
|
|
const candidates = await this.prisma.smsChannel.findMany({ where, select: { id: true, name: true } });
|
|
const total = candidates.length;
|
|
if (total === 0) return { items: [], total, page, pageSize };
|
|
const day = currentShanghaiDayRange();
|
|
const counts = await this.prisma.$queryRaw<Array<{ channelId: string; total: number }>>(Prisma.sql`
|
|
SELECT submit."channelId" AS "channelId", COUNT(*)::integer AS total
|
|
FROM "SmsSubmitRecord" submit
|
|
WHERE submit."channelId" IN (${Prisma.join(candidates.map((channel) => channel.id))})
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") >= ${day.startAt}
|
|
AND COALESCE(submit."submittedAt", submit."createdAt") < ${day.endAt}
|
|
AND submit."submitStatus" IN ('accepted', 'rejected', 'timeout')
|
|
GROUP BY submit."channelId"
|
|
`);
|
|
const countByChannel = new Map(counts.map((row) => [row.channelId, Number(row.total)]));
|
|
// 排序必须发生在分页前，否则只能重排当前页，翻页后会破坏“今日提交量降序”的业务口径。
|
|
const pageIds = candidates
|
|
.sort((left, right) => (countByChannel.get(right.id) ?? 0) - (countByChannel.get(left.id) ?? 0)
|
|
|| left.name.localeCompare(right.name, 'zh-CN')
|
|
|| left.id.localeCompare(right.id))
|
|
.slice((page - 1) * pageSize, page * pageSize)
|
|
.map((channel) => channel.id);
|
|
const pageItems = await this.prisma.smsChannel.findMany({ where: { id: { in: pageIds } }, include: { connectionStates: true } });
|
|
const itemById = new Map(pageItems.map((item) => [item.id, item]));
|
|
const items = pageIds.flatMap((id) => {
|
|
const item = itemById.get(id);
|
|
return item ? [item] : [];
|
|
});
|
|
return { items, total, page, pageSize };
|
|
}
|
|
|
|
async createChannel(data: CreateChannelDto) {
|
|
assertMoneyUnits(data.unitPrice ?? 0, '通道单价');
|
|
const missingFields = ['code', 'name', 'gatewayHost', '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 ?? 7890);
|
|
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,
|
|
data.heartbeatIntervalSeconds,
|
|
data.heartbeatMissThreshold,
|
|
);
|
|
const rateLimitPerSecond = normalizeChannelRateLimit(data.rateLimitPerSecond);
|
|
const carriers = normalizeChannelCarriers(data.carriers, data.carrier);
|
|
const channel = await this.prisma.smsChannel.create({
|
|
data: {
|
|
code: data.code,
|
|
name: data.name,
|
|
carrier: legacyCarrierFromCapabilities(carriers),
|
|
carriers,
|
|
sendRegion: data.sendRegion ?? '全国',
|
|
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.connection.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');
|
|
}
|
|
if (data.unitPrice !== undefined) {
|
|
assertMoneyUnits(data.unitPrice, '通道单价');
|
|
}
|
|
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
|
|
|| data.heartbeatIntervalSeconds !== undefined
|
|
|| data.heartbeatMissThreshold !== undefined
|
|
? normalizeChannelRuntimeConfig(
|
|
channel.config,
|
|
data.config,
|
|
data.desiredConnections,
|
|
data.windowSize,
|
|
data.heartbeatIntervalSeconds,
|
|
data.heartbeatMissThreshold,
|
|
)
|
|
: undefined;
|
|
const rateLimitPerSecond = data.rateLimitPerSecond === undefined
|
|
? undefined
|
|
: normalizeChannelRateLimit(data.rateLimitPerSecond);
|
|
const existingCarriers = normalizeChannelCarriers(channel.carriers, channel.carrier);
|
|
const carriers = data.carriers !== undefined || data.carrier !== undefined
|
|
? normalizeChannelCarriers(data.carriers, data.carrier)
|
|
: existingCarriers;
|
|
if (data.carriers !== undefined || data.carrier !== undefined) {
|
|
const removed = existingCarriers.filter((carrier) => !carriers.includes(carrier));
|
|
if (removed.length) {
|
|
const blockingGroups = await this.prisma.smsChannelGroupItem.findMany({
|
|
where: { channelId, group: { status: 'active', carrier: { in: removed } } },
|
|
include: { group: true },
|
|
});
|
|
if (blockingGroups.length) {
|
|
throw new BadRequestException(`请先解除以下活动通道组引用：${blockingGroups.map((item) => item.group.name).join('、')}`);
|
|
}
|
|
}
|
|
}
|
|
const connectionConfigChanged = channelConnectionSettingsChanged(channel, {
|
|
gatewayHost: data.gatewayHost ?? channel.gatewayHost,
|
|
gatewayPort: gatewayPort ?? channel.gatewayPort,
|
|
account: data.account ?? channel.account,
|
|
passwordCipher: data.passwordCipher ?? channel.passwordCipher,
|
|
cmppVersion: cmppVersion ?? channel.cmppVersion,
|
|
config: config ?? channel.config,
|
|
});
|
|
const updated = await this.prisma.smsChannel.update({
|
|
where: { id: channelId },
|
|
data: {
|
|
code: data.code,
|
|
name: data.name,
|
|
carrier: data.carriers !== undefined || data.carrier !== undefined ? legacyCarrierFromCapabilities(carriers) : undefined,
|
|
carriers: data.carriers !== undefined || data.carrier !== undefined ? carriers : undefined,
|
|
sendRegion: data.sendRegion,
|
|
protocol: 'CMPP',
|
|
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,
|
|
carriers: channel.carriers,
|
|
sendRegion: channel.sendRegion,
|
|
gatewayHost: channel.gatewayHost,
|
|
gatewayPort: channel.gatewayPort,
|
|
enterpriseCode: channel.enterpriseCode,
|
|
account: channel.account,
|
|
srcId: channel.srcId,
|
|
unitPrice: moneyToNumber(channel.unitPrice),
|
|
},
|
|
after: data,
|
|
} as Prisma.InputJsonValue,
|
|
},
|
|
});
|
|
const updatedStatus = data.status ?? channel.status;
|
|
if (updatedStatus === 'active' && (connectionConfigChanged || channel.status !== 'active')) {
|
|
await this.connection.requestChannelConnection(updated, 'channel_updated');
|
|
} else if (updatedStatus !== 'active' && channel.status === 'active') {
|
|
await this.connection.requestChannelDisconnection(updated, 'channel_disabled');
|
|
}
|
|
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.connection.requestChannelConnection(updated, 'channel_enabled', data.operatorId);
|
|
} else if (channel.status === 'active' || data.status === 'deleted') {
|
|
await this.connection.requestChannelDisconnection(
|
|
updated,
|
|
data.status === 'deleted' ? 'channel_deleted' : 'channel_disabled',
|
|
data.operatorId,
|
|
);
|
|
}
|
|
return updated;
|
|
}
|
|
}
|