Event Processing with Redis Streams

Redis Streams are append-only logs with consumer groups. They sit between simple Redis lists and full message brokers like Kafka: persistent, replayable and easy to run.

Producing events

await redis.xAdd('orders', '*', {        // '*' = auto-generated id (timestamp-sequence)
  orderId: '42',
  customerId: '7',
  total: '1499',
});

Consumer groups

A consumer group lets several workers share a stream: each message goes to one worker in the group, and different groups each get every message.

// Create once ('$' = only new messages; '0' = from the beginning)
await redis.xGroupCreate('orders', 'billing', '$', { MKSTREAM: true }).catch(() => {});

while (true) {
  const response = await redis.xReadGroup('billing', `worker-${process.pid}`,
    { key: 'orders', id: '>' },               // '>' = messages never delivered to this group
    { COUNT: 10, BLOCK: 5000 });

  for (const stream of response ?? []) {
    for (const message of stream.messages) {
      await chargeCustomer(message.message);
      await redis.xAck('orders', 'billing', message.id);   // done
    }
  }
}

Handling failures

A message delivered but never acknowledged (because the worker crashed) stays in the group's pending entries list. Claim old pending messages and retry them:

const claimed = await redis.xAutoClaim('orders', 'billing', `worker-${process.pid}`, 60_000, '0-0', { COUNT: 10 });
// messages idle for more than 60 seconds are now ours to process

Track delivery counts with XPENDING and move messages that keep failing to a dead-letter stream.

Keep memory in check

await redis.xAdd('orders', '*', fields, { TRIM: { strategy: 'MAXLEN', strategyModifier: '~', threshold: 100000 } });

Streams or a broker?

• Redis Streams: you already run Redis, volumes fit in memory, and you want simplicity.
• Kafka and similar: very high throughput, long retention on disk, many independent consumers, stream processing ecosystems.