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

fix: harden real backend workflows and channel connections

Browse Source
This commit is contained in:
hectorzhao
2026-07-06 17:54:53 +08:00
parent 8cca361441
commit b5132d7f4e
47 changed files with 2530 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/channels/channels.service.ts
+326 -16
View File
@@ -1,5 +1,7 @@
import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Queue } from 'bullmq';
import { Prisma } from '@prisma/client';
import { randomUUID } from 'crypto';
import { PrismaService } from '../prisma/prisma.service';
export interface CreateChannelDto {
@@ -113,6 +115,7 @@ export interface CreateReceiptImportDto {
export interface UpsertConnectionStateDto {
tenantId?: string;
applicationId?: string;
channelId: string;
connectionId: string;
status: string;
@@ -137,15 +140,45 @@ export interface CopyChannelDto {
operatorId?: string;
}
const GATEWAY_CONNECTION_QUEUE = 'gateway.connection.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';
@Injectable()
export class ChannelsService {
export class ChannelsService implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(ChannelsService.name);
private gatewayConnectionQueue?: Queue;
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();
}
listChannels() {
return this.prisma.smsChannel.findMany({ orderBy: { createdAt: 'desc' }, take: 100 });
}
createChannel(data: CreateChannelDto) {
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 === '';
@@ -157,7 +190,7 @@ export class ChannelsService {
if (!Number.isInteger(gatewayPort) || gatewayPort <= 0 || gatewayPort > 65535) {
throw new BadRequestException('gatewayPort must be an integer between 1 and 65535');
}
return this.prisma.smsChannel.create({
const channel = await this.prisma.smsChannel.create({
data: {
code: data.code,
name: data.name,
@@ -177,6 +210,10 @@ export class ChannelsService {
config: data.config as Prisma.InputJsonValue | undefined,
},
});
if (channel.status === 'active') {
await this.requestChannelConnection(channel, 'channel_created');
}
return channel;
}
async updateChannel(channelId: string, data: UpdateChannelDto) {
@@ -253,6 +290,9 @@ export class ChannelsService {
} as Prisma.InputJsonValue,
},
});
if (data.status === 'active') {
await this.requestChannelConnection(updated, 'channel_enabled', data.operatorId);
}
return updated;
}
@@ -366,7 +406,7 @@ export class ChannelsService {
});
}
async listChannelLinkLogs(channelId: string) {
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');
@@ -412,34 +452,54 @@ export class ChannelsService {
}
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,
status: data.status,
applicationId: data.applicationId,
status,
desiredConnections: data.desiredConnections ?? 1,
currentConnections: data.currentConnections ?? (data.status === 'online' || data.status === 'connected' ? 1 : 0),
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: data.lastError,
};
const state = await this.prisma.cmppConnectionState.upsert({
where: { channelId_connectionId: { channelId: data.channelId, connectionId: data.connectionId } },
update: payload,
create: {
const existing = await this.prisma.cmppConnectionState.findFirst({
where: {
applicationId: data.applicationId ?? null,
channelId: data.channelId,
connectionId: data.connectionId,
...payload,
},
});
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(data.status)}`,
action: `cmpp_connection.${normalizeConnectionAction(status)}`,
resource: 'cmpp_connection',
resourceId: `${data.channelId}:${data.connectionId}`,
detail: {
status: data.status,
status,
applicationId: state.applicationId,
desiredConnections: state.desiredConnections,
currentConnections: state.currentConnections,
lastError: state.lastError,
@@ -449,6 +509,69 @@ export class ChannelsService {
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: true }, orderBy: [{ province: 'asc' }, { priority: 'asc' }] } },
@@ -582,6 +705,25 @@ export class ChannelsService {
});
}
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 },
@@ -796,11 +938,116 @@ export class ChannelsService {
},
});
}
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 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 normalizeConnectionAction(status: string) {
const normalized = status.toLowerCase();
if (['online', 'connected', 'open'].includes(normalized)) {
if (normalized === 'connected') {
return 'connected';
}
if (['heartbeat', 'active_test'].includes(normalized)) {
@@ -812,9 +1059,66 @@ function normalizeConnectionAction(status: string) {
if (['offline', 'closed', 'disconnected'].includes(normalized)) {
return 'disconnected';
}
if (['auth_failed', 'heartbeat_timeout', 'failed', 'error'].includes(normalized)) {
return 'failed';
}
return 'updated';
}
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 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) {
@@ -990,8 +1294,11 @@ function validateGroupItems(
}
function normalizeLinkEvent(action: string) {
if (action.includes('connect_requested')) {
return '连接请求';
}
if (action.includes('connected')) {
return '新建';
return '连接成功';
}
if (action.includes('heartbeat')) {
return '心跳';
@@ -1002,6 +1309,9 @@ function normalizeLinkEvent(action: string) {
if (action.includes('disconnected')) {
return '断开';
}
if (action.includes('failed')) {
return '连接失败';
}
if (action.includes('copy')) {
return '复制';
}