fix(reload): prevent duplicate/destructive reload during transitions

src/defaults.ts

@@ -108,6 +108,8 @@ export const UNOAPI_QUEUE_BULK_WEBHOOK = `${UNOAPI_QUEUE_NAME}.bulk.webhook`
 export const UNOAPI_QUEUE_COMMANDER = `${UNOAPI_QUEUE_NAME}.commander`
 export const UNOAPI_QUEUE_INCOMING = `${UNOAPI_QUEUE_NAME}.incoming`
 export const UNOAPI_QUEUE_TRANSCRIBER = `${UNOAPI_QUEUE_NAME}.transcribe`
+export const RELOAD_PUBLISH_BROKER = process.env.RELOAD_PUBLISH_BROKER === _undefined ? false : process.env.RELOAD_PUBLISH_BROKER == 'true'
+export const RELOAD_BAILEYS_DEBOUNCE_MS = parseInt(process.env.RELOAD_BAILEYS_DEBOUNCE_MS || '15000')
 export const UNOAPI_QUEUE_SPEECH = `${UNOAPI_QUEUE_NAME}.speech`
 export const UNOAPI_MESSAGE_RETRY_LIMIT = parseInt(process.env.UNOAPI_MESSAGE_RETRY_LIMIT || '5')
 export const UNOAPI_MESSAGE_RETRY_DELAY = parseInt(process.env.UNOAPI_MESSAGE_RETRY_DELAY || '10000')

src/services/reload_amqp.ts

@@ -1,5 +1,5 @@
 import { amqpPublish } from '../amqp'
-import { UNOAPI_EXCHANGE_BRIDGE_NAME, UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_RELOAD } from '../defaults'
+import { RELOAD_PUBLISH_BROKER, UNOAPI_EXCHANGE_BRIDGE_NAME, UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_RELOAD } from '../defaults'
 import { getConfig } from './config'
 import { Reload } from './reload'
 
@@ -13,7 +13,21 @@ export class ReloadAmqp extends Reload {
 
   public async run(phone: string) {
     const config = await this.getConfig(phone)
-    await amqpPublish(UNOAPI_EXCHANGE_BROKER_NAME, UNOAPI_QUEUE_RELOAD, phone, { phone }, { type: 'topic' })
-    await amqpPublish(UNOAPI_EXCHANGE_BRIDGE_NAME, `${UNOAPI_QUEUE_RELOAD}.${config.server!}`, '', { phone }, { type: 'direct' })
+    if (RELOAD_PUBLISH_BROKER) {
+      await amqpPublish(
+        UNOAPI_EXCHANGE_BROKER_NAME,
+        UNOAPI_QUEUE_RELOAD,
+        phone,
+        { phone },
+        { type: 'topic' }
+      )
+    }
+    await amqpPublish(
+      UNOAPI_EXCHANGE_BRIDGE_NAME,
+      `${UNOAPI_QUEUE_RELOAD}.${config.server!}`,
+      '',
+      { phone },
+      { type: 'direct' }
+    )
   }
 }

src/services/reload_baileys.ts

@@ -1,4 +1,4 @@
-import { UNOAPI_SERVER_NAME } from '../defaults'
+import { RELOAD_BAILEYS_DEBOUNCE_MS, UNOAPI_SERVER_NAME } from '../defaults'
 import { getClient } from '../services/client'
 import { getConfig } from '../services/config'
 import { Listener } from '../services/listener'
@@ -7,6 +7,8 @@ import logger from './logger'
 import { Reload } from './reload'
 
 export class ReloadBaileys extends Reload {
+  private static readonly inFlightByPhone: Set<string> = new Set()
+  private static readonly lastRunAtByPhone: Map<string, number> = new Map()
   private getClient: getClient
   private getConfig: getConfig
   private listener: Listener
@@ -21,6 +23,16 @@ export class ReloadBaileys extends Reload {
   }
 
   async run(phone: string) {
+    const now = Date.now()
+    const lastRunAt = ReloadBaileys.lastRunAtByPhone.get(phone) || 0
+    const debounceRemaining = Math.max(0, RELOAD_BAILEYS_DEBOUNCE_MS - (now - lastRunAt))
+    if (ReloadBaileys.inFlightByPhone.has(phone) || debounceRemaining > 0) {
+      logger.warn('Skip duplicated reload for %s (inFlight=%s debounceRemainingMs=%s)', phone, ReloadBaileys.inFlightByPhone.has(phone), debounceRemaining)
+      return
+    }
+    ReloadBaileys.inFlightByPhone.add(phone)
+    ReloadBaileys.lastRunAtByPhone.set(phone, now)
+    try {
     logger.debug('Reload baileys run for phone %s', phone)
     const config = await this.getConfig(phone)
     if (config.server != UNOAPI_SERVER_NAME) {
@@ -35,19 +47,28 @@ export class ReloadBaileys extends Reload {
     })
     const store = await config.getStore(phone, config)
     const { sessionStore } = store
-    if ((await sessionStore.isStatusOnline(phone)) || (await sessionStore.isStatusStandBy(phone)) || (await sessionStore.isStatusConnecting(phone))) {
+    const isConnecting = await sessionStore.isStatusConnecting(phone)
+    const isRestartRequired = await sessionStore.isStatusRestartRequired(phone)
+    if (isConnecting || isRestartRequired) {
+      logger.warn('Skip destructive reload for %s while status is transitional (connecting=%s restartRequired=%s)', phone, isConnecting, isRestartRequired)
+      return
+    }
+    const isOnline = await sessionStore.isStatusOnline(phone)
+    const isStandBy = await sessionStore.isStatusStandBy(phone)
+    if (isOnline || isStandBy) {
       logger.warn('Reload disconnect session %s!', phone)
-      await currentClient?.disconnect()
+      await currentClient.disconnect()
     }
     await super.run(phone)
-    await sessionStore.setStatus(phone, 'online') // to clear standby
-    await sessionStore.setStatus(phone, 'disconnected')
     await this.getClient({
       phone,
       listener: this.listener,
       getConfig: this.getConfig,
       onNewLogin: this.onNewLogin,
     })
     logger.info('Reloaded session %s!', phone)
+    } finally {
+      ReloadBaileys.inFlightByPhone.delete(phone)
+    }
   }
 }