[core] Fix ack for status@broadcast

fix #329

package.json

@@ -35,6 +35,7 @@
     "@nestjs/terminus": "^10.2.3",
     "@types/lodash": "^4.14.194",
     "@types/ws": "^8.5.4",
+    "async-lock": "^1.4.1",
     "check-disk-space": "^3.4.0",
     "class-validator": "^0.12.2",
     "del": "^6.0.0",

src/config.service.ts

@@ -112,17 +112,6 @@ export class WhatsappConfigService {
     return this.configService.get('WHATSAPP_SESSIONS_MONGO_URL', undefined);
   }
 
-  getDefaultEngineName(): WAHAEngine {
-    const value = getEngineName();
-    if (value in WAHAEngine) {
-      return WAHAEngine[value];
-    }
-    console.log(
-      `Unknown WhatsApp default engine WHATSAPP_DEFAULT_ENGINE=${value}. Using WEBJS`,
-    );
-    return WAHAEngine.WEBJS;
-  }
-
   get(name: string, defaultValue: any = undefined): any {
     return this.configService.get(name, defaultValue);
   }

src/core/config/EngineConfigService.ts

@@ -2,11 +2,24 @@ import { Injectable } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 
 import { parseBool } from '../../helpers';
+import { WAHAEngine } from '../../structures/enums.dto';
+import { getEngineName } from '../../version';
 
 @Injectable()
 export class EngineConfigService {
   constructor(protected configService: ConfigService) {}
 
+  getDefaultEngineName(): WAHAEngine {
+    const value = getEngineName();
+    if (value in WAHAEngine) {
+      return WAHAEngine[value];
+    }
+    console.log(
+      `Unknown WhatsApp default engine WHATSAPP_DEFAULT_ENGINE=${value}. Using WEBJS`,
+    );
+    return WAHAEngine.WEBJS;
+  }
+
   get shouldPrintQR(): boolean {
     const value = this.configService.get('WAHA_PRINT_QR', true);
     return parseBool(value);

src/core/engines/noweb/NowebAuthFactoryCore.ts

@@ -1,8 +1,6 @@
-import { useMultiFileAuthState } from '@adiwajshing/baileys';
-import { Store } from 'whatsapp-web.js';
-
 import { DataStore } from '../../abc/DataStore';
 import { LocalStore } from '../../storage/LocalStore';
+import { useMultiFileAuthState } from './useMultiFileAuthState';
 
 export class NowebAuthFactoryCore {
   buildAuth(store: DataStore, name: string) {

src/core/engines/noweb/session.noweb.core.ts

@@ -5,6 +5,7 @@ import makeWASocket, {
   getAggregateVotesInPollMessage,
   getKeyAuthor,
   isJidGroup,
+  isJidStatusBroadcast,
   jidNormalizedUser,
   makeCacheableSignalKeyStore,
   makeInMemoryStore,
@@ -109,9 +110,13 @@ const PresenceStatuses = {
 const ToEnginePresenceStatus = flipObject(PresenceStatuses);
 
 export class WhatsappSessionNoWebCore extends WhatsappSession {
+  private START_ATTEMPT_DELAY_SECONDS = 2;
+  private AUTO_RESTART_AFTER_SECONDS = 30 * 60;
+
   engine = WAHAEngine.NOWEB;
   authFactory = new NowebAuthFactoryCore();
-  private restartTimeoutId: null | ReturnType<typeof setTimeout> = null;
+  private startTimeoutId: null | ReturnType<typeof setTimeout> = null;
+  private autoRestartTimeoutId: null | ReturnType<typeof setTimeout> = null;
 
   get listenConnectionEventsFromTheStart() {
     return true;
@@ -194,6 +199,23 @@ export class WhatsappSessionNoWebCore extends WhatsappSession {
       this.listenConnectionEvents();
       this.events.emit(WAHAInternalEvent.ENGINE_START);
     }
+    this.enableAutoRestart();
+  }
+
+  protected enableAutoRestart() {
+    if (this.autoRestartTimeoutId) {
+      return;
+    }
+
+    // add random delay to avoid multiple restarts at the same time
+    const shiftSeconds = Math.floor(Math.random() * 30);
+    const delay = this.AUTO_RESTART_AFTER_SECONDS + shiftSeconds;
+    this.log.debug(`Auto-restart is enabled, after ${delay} seconds`);
+    this.autoRestartTimeoutId = setTimeout(() => {
+      this.autoRestartTimeoutId = null;
+      this.log.log('Auto-restarting the client connection...');
+      this.sock?.end('auto-restart');
+    }, delay * SECOND);
   }
 
   protected async getMessage(
@@ -213,17 +235,19 @@ export class WhatsappSessionNoWebCore extends WhatsappSession {
   }
 
   private restartClient() {
-    if (this.restartTimeoutId) {
+    if (this.startTimeoutId) {
       this.log.log(
         'Request to restart is already in progress, ignoring this request',
       );
       return;
     }
-    this.log.log('Setting up client restart in 2 seconds...');
-    this.restartTimeoutId = setTimeout(() => {
-      this.restartTimeoutId = undefined;
+    this.log.log(
+      `Setting up client start in ${this.START_ATTEMPT_DELAY_SECONDS} seconds...`,
+    );
+    this.startTimeoutId = setTimeout(() => {
+      this.startTimeoutId = undefined;
       this.buildClient();
-    }, 2 * SECOND);
+    }, this.START_ATTEMPT_DELAY_SECONDS * SECOND);
   }
 
   protected listenConnectionEvents() {
@@ -241,17 +265,18 @@ export class WhatsappSessionNoWebCore extends WhatsappSession {
         const shouldReconnect =
           lastDisconnect.error?.output?.statusCode !==
           DisconnectReason.loggedOut;
-        this.log.error(
-          'connection closed due to ',
-          lastDisconnect.error,
-          ', reconnecting ',
-          shouldReconnect,
-        );
         this.qr.save('');
         // reconnect if not logged out
         if (shouldReconnect) {
+          this.log.log(
+            `Connection closed due to '${lastDisconnect.error}', reconnecting...`,
+          );
           this.restartClient();
         } else {
+          this.log.error(
+            `Connection closed due to '${lastDisconnect.error}', do not reconnect the session.`,
+          );
+          await this.end();
           this.status = WAHASessionStatus.FAILED;
         }
       }
@@ -267,13 +292,19 @@ export class WhatsappSessionNoWebCore extends WhatsappSession {
   }
 
   async stop() {
-    this.sock?.ev?.removeAllListeners();
-    this.sock?.ws?.removeAllListeners();
-    this.sock?.ws?.close();
+    await this.end();
     this.status = WAHASessionStatus.STOPPED;
     return;
   }
 
+  private async end() {
+    clearTimeout(this.autoRestartTimeoutId);
+    clearTimeout(this.startTimeoutId);
+    this.sock?.ev?.removeAllListeners();
+    this.sock?.ws?.removeAllListeners();
+    this.sock?.end();
+  }
+
   async getSessionMeInfo(): Promise<MeInfo | null> {
     const me = this.sock.authState?.creds?.me;
     if (!me) {
@@ -980,6 +1011,9 @@ function toCusFormat(remoteJid) {
   if (isJidGroup(remoteJid)) {
     return remoteJid;
   }
+  if (isJidStatusBroadcast(remoteJid)) {
+    return remoteJid;
+  }
   if (!remoteJid) {
     return;
   }

src/core/engines/noweb/useMultiFileAuthState.ts

@@ -0,0 +1,115 @@
+import {
+  AuthenticationCreds,
+  AuthenticationState,
+  proto,
+} from '@adiwajshing/baileys';
+import { BufferJSON, initAuthCreds } from '@adiwajshing/baileys/lib/Utils';
+import { mkdir, readFile, stat, unlink, writeFile } from 'fs/promises';
+import { join } from 'path';
+
+// eslint-disable-next-line @typescript-eslint/no-var-requires
+const AsyncLock = require('async-lock');
+
+// We need to lock files due to the fact that we are using async functions to read and write files
+// https://github.com/WhiskeySockets/Baileys/issues/794
+// https://github.com/nodejs/node/issues/26338
+// Default pending is 1000, set it to infinity
+// https://github.com/rogierschouten/async-lock/issues/63
+const fileLock = new AsyncLock({ maxPending: Infinity });
+
+/**
+ * stores the full authentication state in a single folder.
+ * Far more efficient than singlefileauthstate
+ *
+ * Again, I wouldn't endorse this for any production level use other than perhaps a bot.
+ * Would recommend writing an auth state for use with a proper SQL or No-SQL DB
+ * */
+export const useMultiFileAuthState = async (
+  folder: string,
+): Promise<{
+  state: AuthenticationState;
+  saveCreds: () => Promise<void>;
+}> => {
+  const writeData = (data: any, file: string) => {
+    const filePath = join(folder, fixFileName(file));
+    return fileLock.acquire(filePath, () =>
+      writeFile(join(filePath), JSON.stringify(data, BufferJSON.replacer)),
+    );
+  };
+
+  const readData = async (file: string) => {
+    try {
+      const filePath = join(folder, fixFileName(file));
+      const data = await fileLock.acquire(filePath, () =>
+        readFile(filePath, { encoding: 'utf-8' }),
+      );
+      return JSON.parse(data, BufferJSON.reviver);
+    } catch (error) {
+      return null;
+    }
+  };
+
+  const removeData = async (file: string) => {
+    try {
+      const filePath = join(folder, fixFileName(file));
+      await fileLock.acquire(filePath, () => unlink(filePath));
+    } catch {}
+  };
+
+  const folderInfo = await stat(folder).catch(() => {
+    return null;
+  });
+  if (folderInfo) {
+    if (!folderInfo.isDirectory()) {
+      throw new Error(
+        `found something that is not a directory at ${folder}, either delete it or specify a different location`,
+      );
+    }
+  } else {
+    await mkdir(folder, { recursive: true });
+  }
+
+  const fixFileName = (file?: string) =>
+    file?.replace(/\//g, '__')?.replace(/:/g, '-') || '';
+
+  const creds: AuthenticationCreds =
+    (await readData('creds.json')) || initAuthCreds();
+
+  return {
+    state: {
+      creds,
+      keys: {
+        get: async (type, ids) => {
+          const data = {};
+          await Promise.all(
+            ids.map(async (id) => {
+              let value = await readData(`${type}-${id}.json`);
+              if (type === 'app-state-sync-key' && value) {
+                value = proto.Message.AppStateSyncKeyData.fromObject(value);
+              }
+
+              data[id] = value;
+            }),
+          );
+
+          return data;
+        },
+        set: async (data) => {
+          const tasks: Promise<void>[] = [];
+          for (const category in data) {
+            for (const id in data[category]) {
+              const value = data[category][id];
+              const file = `${category}-${id}.json`;
+              tasks.push(value ? writeData(value, file) : removeData(file));
+            }
+          }
+
+          await Promise.all(tasks);
+        },
+      },
+    },
+    saveCreds: () => {
+      return writeData(creds, 'creds.json');
+    },
+  };
+};

src/core/engines/webjs/session.webjs.core.ts

@@ -31,6 +31,7 @@ import {
 import { ContactQuery, ContactRequest } from '../../../structures/contacts.dto';
 import {
   ACK_UNKNOWN,
+  SECOND,
   WAHAEngine,
   WAHAEvents,
   WAHAPresenceStatus,
@@ -67,15 +68,21 @@ export interface WebJSConfig {
 }
 
 export class WhatsappSessionWebJSCore extends WhatsappSession {
+  private START_ATTEMPT_DELAY_SECONDS = 2;
+
   engine = WAHAEngine.WEBJS;
   protected engineConfig?: WebJSConfig;
 
+  private startTimeoutId: null | ReturnType<typeof setTimeout> = null;
+  private shouldRestart: boolean;
+
   whatsapp: Client;
   protected qr: QR;
 
   public constructor(config) {
     super(config);
     this.qr = new QR();
+    this.shouldRestart = true;
   }
 
   /**
@@ -94,6 +101,7 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
         headless: true,
         executablePath: this.getBrowserExecutablePath(),
         args: this.getBrowserArgsForPuppeteer(),
+        dumpio: this.isDebugEnabled(),
       },
       webVersion: webVersion,
       webVersionCache: {
@@ -110,6 +118,27 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
     return new Client(clientOptions);
   }
 
+  private restartClient() {
+    if (!this.shouldRestart) {
+      this.log.debug("Shouldn't restart the client, ignoring restart request");
+      return;
+    }
+
+    if (this.startTimeoutId) {
+      const msg =
+        'Request to restart is already in progress, ignoring restart request';
+      this.log.warn(msg);
+      return;
+    }
+    this.log.log(
+      `Setting up client start in ${this.START_ATTEMPT_DELAY_SECONDS} seconds...`,
+    );
+    this.startTimeoutId = setTimeout(() => {
+      this.startTimeoutId = undefined;
+      this.start();
+    }, this.START_ATTEMPT_DELAY_SECONDS * SECOND);
+  }
+
   protected addProxyConfig(clientOptions: ClientOptions) {
     if (this.proxyConfig?.server !== undefined) {
       // push the proxy server to the args
@@ -127,8 +156,9 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
     }
   }
 
-  async start() {
-    this.status = WAHASessionStatus.STARTING;
+  protected async init() {
+    await this.end();
+    this.shouldRestart = true;
     this.whatsapp = await this.buildClient();
     this.whatsapp
       .initialize()
@@ -137,12 +167,16 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
         this.whatsapp.pupBrowser.on('disconnected', () => {
           this.status = WAHASessionStatus.FAILED;
           this.log.error('The browser has been disconnected');
+          this.restartClient();
         });
         // Listen for page close event
         this.whatsapp.pupPage.on('close', () => {
           this.status = WAHASessionStatus.FAILED;
           this.log.error('The WhatsApp Web page has been closed');
+          this.restartClient();
         });
+
+        // Listen for page error event
         if (this.isDebugEnabled()) {
           this.log.debug("Logging 'console' event for web page");
           this.whatsapp.pupPage.on('console', (msg) =>
@@ -156,6 +190,7 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
       .catch((error) => {
         this.status = WAHASessionStatus.FAILED;
         this.log.error(error);
+        this.restartClient();
         return;
       });
     if (this.isDebugEnabled()) {
@@ -163,12 +198,30 @@ export class WhatsappSessionWebJSCore extends WhatsappSession {
     }
     this.listenConnectionEvents();
     this.events.emit(WAHAInternalEvent.ENGINE_START);
+  }
+
+  async start() {
+    this.status = WAHASessionStatus.STARTING;
+    await this.init();
     return this;
   }
 
   async stop() {
-    await this.whatsapp.destroy();
+    this.shouldRestart = false;
     this.status = WAHASessionStatus.STOPPED;
+    await this.end();
+  }
+
+  private async end() {
+    try {
+      this.whatsapp?.removeAllListeners();
+      clearInterval(this.startTimeoutId);
+      this.whatsapp?.destroy().catch((error) => {
+        this.log.debug('Failed to destroy the client', error);
+      });
+    } catch (error) {
+      this.log.error(error);
+    }
   }
 
   async getSessionMeInfo(): Promise<MeInfo | null> {

src/core/manager.core.ts

@@ -62,7 +62,7 @@ export class SessionManagerCore extends SessionManager {
 
     this.log.setContext('SessionManager');
     this.session = undefined;
-    const engineName = this.config.getDefaultEngineName();
+    const engineName = this.engineConfigService.getDefaultEngineName();
     this.EngineClass = this.getEngine(engineName);
     this.store = new LocalStoreCore(engineName.toLowerCase());
     this.sessionAuthRepository = new LocalSessionAuthRepository(this.store);

src/version.ts

@@ -33,7 +33,7 @@ export function getEngineName(): string {
 }
 
 export const VERSION: WAHAEnvironment = {
-  version: '2024.5.13',
+  version: '2024.5.14',
   engine: getEngineName(),
   tier: getWAHAVersion(),
   browser: getBrowserExecutablePath(),

yarn.lock

@@ -14,10 +14,11 @@ __metadata:
 
 "@adiwajshing/baileys@github:WhiskeySockets/Baileys":
   version: 6.7.2
-  resolution: "@adiwajshing/baileys@https://github.com/WhiskeySockets/Baileys.git#commit=79c33e53b4124735f019c48e8075c9cde807b7e7"
+  resolution: "@adiwajshing/baileys@https://github.com/WhiskeySockets/Baileys.git#commit=9065ab690fde0641e3eac458c1521e1d151c5050"
   dependencies:
     "@adiwajshing/keyed-db": ^0.2.4
     "@hapi/boom": ^9.1.3
+    async-lock: ^1.4.1
     audio-decode: ^2.1.3
     axios: ^1.3.3
     cache-manager: 4.0.1
@@ -44,7 +45,7 @@ __metadata:
       optional: true
     sharp:
       optional: true
-  checksum: dacfec2d2c4767370f2ee9a846e119cedf22d792ae920c5682e66aa67e788ec409d5ee364e6879c7c9e76d00acfb2c26c197d377893aaccc2100f3ae22940d72
+  checksum: 255c9aa3107a56055b33adbfba5cce41bad16d1b2453a19f3896438a85e191786c144ea73ebdb896f58c7436044aca8e740d5d9e7271c8df12f9e89318abf90f
   languageName: node
   linkType: hard
 
@@ -3169,6 +3170,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"async-lock@npm:^1.4.1":
+  version: 1.4.1
+  resolution: "async-lock@npm:1.4.1"
+  checksum: 29e70cd892932b7c202437786cedc39ff62123cb6941014739bd3cabd6106326416e9e7c21285a5d1dc042cad239a0f7ec9c44658491ee4a615fd36a21c1d10a
+  languageName: node
+  linkType: hard
+
 "async@npm:3.2.3":
   version: 3.2.3
   resolution: "async@npm:3.2.3"
@@ -12544,6 +12552,7 @@ __metadata:
     "@types/ws": ^8.5.4
     "@typescript-eslint/eslint-plugin": 3.9.1
     "@typescript-eslint/parser": 3.9.1
+    async-lock: ^1.4.1
     check-disk-space: ^3.4.0
     class-validator: ^0.12.2
     del: ^6.0.0