Message Buffer System with Redis for Efficient Processing

🚀 Message-Batching Buffer Workflow (n8n)

This workflow implements a lightweight message-batching buffer using Redis for temporary storage and a JavaScript consolidation function to merge messages. It collects incoming user messages per session, waits for a configurable inactivity window or batch size threshold, consolidates buffered messages via custom code, then clears the buffer and returns the combined response—all without external LLM calls.

🔑 Key Features

• Redis-backed buffer queues incoming messages for context_id .
• Centralized Config Parameters node to adjust thresholds and timeouts in one place.
• Dynamic wait time based on message length (configurable minWords , waitLong , waitShort ).
• Batch trigger fires on inactivity timeout or when buffer_count ≥ batchThreshold .
• Zero-cost consolidation via built-in JavaScript Function ( consolidate buffer )—no GPT-4 or external API required.

⚙️ Setup Instructions

1. Extract Session & Message

   • Trigger: When chat message received (webhook) or When clicking 'Test workflow' (manual).
   • Map inputs: set variables context_id and message into a Set node named Mock input data (for testing) or a proper mapping node in production.

2. Config Parameters

   • Add a Set node Config Parameters with:

      minWords: 3 # Word threshold
      waitLong: 10 # Timeout (s) for long messages
      waitShort: 20 # Timeout (s) for short messages 
     batchThreshold: 3 # Messages to trigger batch early
     

   • All downstream nodes reference these JSON values ​​dynamically.

3. Determine Wait Time

   • Node: get wait seconds (Code)

   • JS code:

      const msg = $json.message || '';
      const wordCount = msg.split(/\s+/).filter(w => w).length;
      const { minWords, waitLong, waitShort } = items[0].json;
      const waitSeconds = wordCount < minWords ? waitShort : waitLong;
      return [{ json: { context_id: $json.context_id, message: msg, waitSeconds } }];
     

4. Buffer Message in Redis

   • Buffer messages : LPUSH buffer_in: {{$json.context_id}} with payload {text, timestamp} .
   • Set buffer_count increment : INCR buffer_count: {{$json.context_id}} with TTL {{$json.waitSeconds + 60}} .
   • Set last_seen : record last_seen: {{$json.context_id}} timestamp with same TTL.

5. Check & Set Waiting Flag

   • Get waiting_reply : if null, Set waiting_reply to true with TTL {{$json.waitSeconds}} ; else exit.

6. Wait for Inactivity

   • WaitSeconds (webhook): pauses for {{$json.waitSeconds}} seconds before batch evaluation.

7. Check Batch Trigger

   • Get last_seen and Get buffer_count .
   • IF (now - last_seen) ≥ waitSeconds * 1000 OR buffer_count ≥ batchThreshold , proceed; else use Wait node to retry.

8. Consolidate Buffer

   • consolidate buffer (Code):

      const j = items[0].json;
      const raw = Array.isArray(j.buffer) ? j.buffer : [];
      const buffer = raw.map(x => { 
     try { return typeof x === 'string' ? JSON.parse(x) : x;
      } catch { return null; }
      }).filter(Boolean);
      buffer.sort((a, b) => new Date(a.timestamp) - new Date(b.timestamp));
      const texts = buffer.map(e => e.text?.trim()).filter(Boolean);
      const unique = [...new Set(texts)];
      const message = unique.join(' ');
      return [{ json: { context_id: j.context_id, message } }];
     

9. Cleanup & Respond

   • Delete Redis keys: buffer_in , buffer_count , waiting_reply , last_seen (for the context_id ).
   • Return consolidated message to the user via your chat integration.

🛠 Customization Guidance

• Adjust thresholds by editing the Config Parameters node.
• Change concatenation (eg, line breaks) by modifying the join separator in the consolidation code.
• Add filters (eg, ignore empty or system messages) inside the consolidation Function.
• Monitor performance : for very high volume, consider sharding Redis keys by date or user segments.

© 2025 Innovatex • Automation & AI Solutions • innovatexiot.carrd.co • LinkedIn