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 per 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 (e.g., line breaks) by modifying the join separator in the consolidation code.
• Add filters (e.g., 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