feat: Optimize Offline Node Processing with Batching and Event Loop Yielding

[Santosl2]

This pull request optimizes the processing of offline message nodes in the makeMessagesRecvSocket function by introducing batching and yielding to the event loop. This prevents the event loop from being blocked when processing a large number of nodes, improving responsiveness and stability.

Event loop optimization:

• Added a yieldToEventLoop helper function that yields control to the event loop using setImmediate, preventing long blocks during node processing.
• Introduced a BATCH_SIZE constant to process nodes in batches and yield after every batch, reducing the risk of blocking the event loop when many offline nodes are present. [1] [2] [3]

[whiskeysockets-bot]

Thanks for opening this pull request and contributing to the project!

The next step is for the maintainers to review your changes. If everything looks good, it will be approved and merged into the main branch.

In the meantime, anyone in the community is encouraged to test this pull request and provide feedback.

✅ How to confirm it works

If you’ve tested this PR, please comment below with:

Tested and working ✅

This helps us speed up the review and merge process.

📦 To test this PR locally:

# NPM
npm install @whiskeysockets/baileys@Santosl2/Baileys-1#feat/improve-offline-processor

# Yarn (v2+)
yarn add @whiskeysockets/baileys@Santosl2/Baileys-1#feat/improve-offline-processor

# PNPM
pnpm add @whiskeysockets/baileys@Santosl2/Baileys-1#feat/improve-offline-processor

If you encounter any issues or have feedback, feel free to comment as well.

[Santosl2]

@jlucaso1 @purpshell please review it and let me know what you think

[Santosl2]

To make it easier, i created this mocked test example. Notice that the second example will execute other tasks while the first one finished EVERYTHING before finally releasing the event loop.
@jlucaso1 @purpshell @canove

const yieldToEventLoop = () => {
  return new Promise((resolve) => setImmediate(resolve));
};
const BATCH_SIZE = 10;

function makeOfflineNodeProcessorOld() {
  const nodes = [];
  let isProcessing = false;
  let processingPromise = null;

  const enqueue = (type, node) => {
    nodes.push({ type, node });

    if (isProcessing) {
      return;
    }

    isProcessing = true;

    processingPromise = (async () => {
      while (nodes.length) {
        const { node } = nodes.shift();

        if (node.id % 100 === 0) {
          console.log(`  Processed node ${node.id}...`);
        }

        await fakeNodeProcessor(node);
      }

      isProcessing = false;
    })();

    processingPromise.catch((error) =>
      console.error("Unexpected error processing offline nodes:", error)
    );
  };

  const waitForCompletion = () => processingPromise || Promise.resolve();

  return { enqueue, waitForCompletion };
}

const fakeNodeProcessor = async (node) => {
  return new Promise((resolve) => {
    let soma = 0;
    for (let j = 0; j < 100000; j++) {
      soma += Math.sqrt(j);
    }
    resolve();
  });
};

function makeOfflineNodeProcessorNew() {
  const nodes = [];
  let isProcessing = false;
  let processingPromise = null;

  const enqueue = (type, node) => {
    nodes.push({ type, node });

    if (isProcessing) {
      return;
    }

    isProcessing = true;

    processingPromise = (async () => {
      let processedInBatch = 0;

      while (nodes.length) {
        const { node } = nodes.shift();

        if (node.id % 100 === 0) {
          console.log(`  Processed node ${node.id}...`);
        }

        await fakeNodeProcessor(node);
        processedInBatch += 1;
        if (processedInBatch >= BATCH_SIZE) {
          processedInBatch = 0;
          await yieldToEventLoop();
        }
      }

      isProcessing = false;
    })();

    processingPromise.catch((error) =>
      console.error("Unexpected error processing offline nodes:", error)
    );
  };

  const waitForCompletion = () => processingPromise || Promise.resolve();

  return { enqueue, waitForCompletion };
}

function runParallelTasks() {
  let timerCount = 0;
  const timer = setInterval(() => {
    timerCount++;
    console.log(`  ⏰ Timer ${timerCount} executed! (should be every 500ms)`);
  }, 500);

  fetch("https://example.com").then(() => {
    console.log("  🌐 Fetched from network! (I/O operation)");
  });

  // Timeout that should execute after 2 seconds
  setTimeout(() => {
    console.log("  ⏰ Timeout of 2s executed!");
  }, 2000);

  // Simulate I/O (file reading)
  const fs = require("fs");
  fs.readFile(__filename, "utf8", (err, data) => {
    console.log("  📁 File read! (I/O operation)");
  });

  return timer;
}

async function processWithImplementation(useOldOrNew = "new") {
  const offlineNodeProcessor = useOldOrNew === "new" ? makeOfflineNodeProcessorNew() : makeOfflineNodeProcessorOld();
  offlineNodeProcessor.enqueue("type", { id: 1 });
  for (let i = 1; i < 10000; i++) {
    offlineNodeProcessor.enqueue("type", { id: i });
  }

  await offlineNodeProcessor.waitForCompletion();
}

async function menu() {
  console.log("╔════════════════════════════════════════════╗");
  console.log("║        BAILEYS OFFLINE NODE PROCESSOR      ║");
  console.log("║          EVENT LOOP TEST NODE.JS           ║");
  console.log("╚════════════════════════════════════════════╝");
  console.log("Choose an option:");
  console.log("1 - Old implementation (will freeze)");
  console.log("2 - Implementation with breaks (won't freeze)");

  const readline = require("readline").createInterface({
    input: process.stdin,
    output: process.stdout,
  });

  readline.question("\nType 1 or 2: ", async (resposta) => {
    readline.close();

    if (resposta === "1") {
      const timer = runParallelTasks();
      await processWithImplementation("old");
      clearInterval(timer);
    } else if (resposta === "2") {
      const timer = runParallelTasks();
      await processWithImplementation();
      clearInterval(timer);
    } else {
      console.log("Invalid option!");
    }

    console.log("\n✅ Program finished!");
    // process.exit(0);
  });
}

menu();

[Salientekill]

Thank you for your work. 🫡👍 I think they're just going to ask you to fix the lint.