278 lines
13 KiB
TypeScript
278 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, isChannelCarrierCompatible, normalizeRegion, isRegionCompatible, validateGroupItems, normalizeReportType, summarizeReportStatuses, normalizeLinkEvent } from './channels.helpers';
|
|
|
|
|
|
/** R5 channel domain service composed behind ChannelsService. */
|
|
export class ChannelGroupRoutingService {
|
|
constructor(private readonly prisma: PrismaService) {}
|
|
|
|
listGroups() {
|
|
return this.prisma.smsChannelGroup.findMany({
|
|
where: { status: { not: 'deleted' } },
|
|
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, channel.carriers)) {
|
|
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,
|
|
},
|
|
});
|
|
}
|
|
|
|
async updateGroup(groupId: string, data: UpdateChannelGroupDto) {
|
|
const current = await this.prisma.smsChannelGroup.findUnique({
|
|
where: { id: groupId },
|
|
include: { items: { include: { channel: true }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
|
|
});
|
|
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,
|
|
})),
|
|
});
|
|
}
|
|
const updated = await tx.smsChannelGroup.findUnique({
|
|
where: { id: groupId },
|
|
include: { items: { include: { channel: true }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
|
|
});
|
|
await tx.operationLog.create({
|
|
data: {
|
|
action: 'sms_channel_group.update',
|
|
resource: 'sms_channel_group',
|
|
resourceId: groupId,
|
|
detail: {
|
|
before: channelGroupAuditSnapshot(current),
|
|
after: updated ? channelGroupAuditSnapshot(updated) : null,
|
|
} as Prisma.InputJsonValue,
|
|
},
|
|
});
|
|
return updated;
|
|
});
|
|
}
|
|
|
|
async getGroupDeletionImpact(groupId: string) {
|
|
const group = await this.prisma.smsChannelGroup.findUnique({
|
|
where: { id: groupId },
|
|
select: { id: true, name: true, items: { select: { id: true } } },
|
|
});
|
|
if (!group) {
|
|
throw new NotFoundException('Channel group not found');
|
|
}
|
|
const routes = await this.prisma.channelRouteRule.findMany({
|
|
where: { groupId, applicationId: { not: null }, status: { not: 'deleted' } },
|
|
select: { applicationId: true },
|
|
});
|
|
const applicationIds = [...new Set(routes.flatMap((route) => route.applicationId ? [route.applicationId] : []))];
|
|
const [applications, pendingSupplierSubmitCount] = await Promise.all([
|
|
this.prisma.smsApplication.findMany({
|
|
where: { id: { in: applicationIds } },
|
|
select: { id: true, status: true },
|
|
}),
|
|
this.prisma.smsSubmitRecord.count({
|
|
where: { channelGroupId: groupId, submitStatus: 'queued' },
|
|
}),
|
|
]);
|
|
const applicationStatusById = new Map(applications.map((application) => [application.id, application.status]));
|
|
const deletedApplicationCount = applicationIds.filter((applicationId) => {
|
|
const status = applicationStatusById.get(applicationId);
|
|
return status === undefined || status === 'deleted';
|
|
}).length;
|
|
|
|
return {
|
|
groupId: group.id,
|
|
groupName: group.name,
|
|
normalApplicationCount: applicationIds.length - deletedApplicationCount,
|
|
deletedApplicationCount,
|
|
channelCount: group.items.length,
|
|
pendingSupplierSubmitCount,
|
|
};
|
|
}
|
|
|
|
async deleteGroup(groupId: string) {
|
|
const group = await this.prisma.smsChannelGroup.findUnique({
|
|
where: { id: groupId },
|
|
include: { items: { include: { channel: true }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
|
|
});
|
|
if (!group) {
|
|
throw new NotFoundException('Channel group not found');
|
|
}
|
|
if (group.status === 'deleted') {
|
|
return group;
|
|
}
|
|
const impact = await this.getGroupDeletionImpact(groupId);
|
|
|
|
// Logical deletion keeps group items and route bindings available for historical
|
|
// receipts and uplink access-number matching; new submits already require an active group.
|
|
return this.prisma.$transaction(async (tx) => {
|
|
const deleted = await tx.smsChannelGroup.update({
|
|
where: { id: groupId },
|
|
data: { status: 'deleted' },
|
|
});
|
|
await tx.operationLog.create({
|
|
data: {
|
|
action: 'sms_channel_group.delete',
|
|
resource: 'sms_channel_group',
|
|
resourceId: groupId,
|
|
detail: {
|
|
before: channelGroupAuditSnapshot(group),
|
|
impact,
|
|
deletionMode: 'soft_delete',
|
|
} as Prisma.InputJsonValue,
|
|
},
|
|
});
|
|
return deleted;
|
|
}, { isolationLevel: Prisma.TransactionIsolationLevel.Serializable });
|
|
}
|
|
|
|
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',
|
|
},
|
|
});
|
|
}
|
|
}
|