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

fix: restore active channels after gateway restart

Browse Source
This commit is contained in:
hectorzhao
2026-07-15 18:30:02 +08:00
parent 7091a8bed4
commit 85ff037647
7 changed files with 61 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/channels/channels.service.ts
+32 -9
View File
@@ -180,6 +180,7 @@ 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 DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS = 1_000;
const CONNECTING_TIMEOUT_ERROR = 'Gateway connection request timed out';
const DEFAULT_CMPP_VERSION = '2.0';
@@ -190,25 +191,32 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
private gatewaySubmitQueue?: Queue;
private redis?: IORedis;
private connectionTimeoutTimer?: ReturnType<typeof setInterval>;
private gatewayStartupReconnectTimer?: ReturnType<typeof setTimeout>;
constructor(private readonly prisma: PrismaService) {}
onModuleInit() {
if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED === 'true') {
return;
if (process.env.GATEWAY_CONNECTING_TIMEOUT_SCANNER_DISABLED !== 'true') {
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?.();
}
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?.();
this.gatewayStartupReconnectTimer = setTimeout(() => {
void this.reconnectActiveChannelsAfterGatewayRestart();
}, getPositiveIntegerEnv('GATEWAY_STARTUP_RECONNECT_DELAY_MS', DEFAULT_GATEWAY_STARTUP_RECONNECT_DELAY_MS));
this.gatewayStartupReconnectTimer.unref?.();
}
async onModuleDestroy() {
if (this.connectionTimeoutTimer) {
clearInterval(this.connectionTimeoutTimer);
}
if (this.gatewayStartupReconnectTimer) {
clearTimeout(this.gatewayStartupReconnectTimer);
}
await this.gatewayConnectionQueue?.close();
await this.gatewaySubmitQueue?.close();
this.redis?.disconnect();
@@ -1232,7 +1240,7 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
rateLimitPerSecond: number;
config?: Prisma.JsonValue | null;
},
reason: 'channel_created' | 'channel_enabled',
reason: 'channel_created' | 'channel_enabled' | 'gateway_restarted',
operatorId?: string,
) {
const desiredConnections = getDesiredConnections(channel.config);
@@ -1300,6 +1308,21 @@ export class ChannelsService implements OnModuleInit, OnModuleDestroy {
return state;
}
private async reconnectActiveChannelsAfterGatewayRestart() {
const channels = await this.prisma.smsChannel.findMany({ where: { status: 'active' } });
const results = await Promise.allSettled(
channels.map((channel) => this.requestChannelConnection(channel, 'gateway_restarted')),
);
results.forEach((result, index) => {
if (result.status === 'rejected') {
const channel = channels[index];
this.logger.error(
`Failed to restore active CMPP channel ${channel?.code ?? channel?.id ?? index}: ${result.reason instanceof Error ? result.reason.message : String(result.reason)}`,
);
}
});
}
private getGatewayConnectionQueue() {
this.gatewayConnectionQueue ??= new Queue(GATEWAY_CONNECTION_QUEUE, { connection: bullmqConnection() });
return this.gatewayConnectionQueue;