The Complete Production SOP: How to Deploy and Run Redis Streams Consumer Groups & Event Pipelines in Autonomous AI Agents

---
name: "redis-streams-consumer-groups"
description: "Implement resilient, horizontally scalable message distribution pipelines using Redis Streams."
metadata:
  author: "Redis"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "create redis consumer group"
  - "read from redis stream xreadgroup"
  - "handle ack redis streams"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Redis Streams Consumer Groups & Event Pipelines

Builds event-driven architectures with Redis Streams: XADD event publishing, consumer groups with XREADGROUP, pending entry tracking (XPENDING), message claiming (XCLAIM), and auto-trimming (MAXLEN).

## When to Invoke
Invoke this skill whenever you need to:
- create redis consumer group
- read from redis stream xreadgroup
- handle ack redis streams

## Execution SOP & Rules
1. Publish stream entry: XADD telemetry:events MAXLEN ~ 100000 * event_type 'click' payload '{...}'.
2. Create consumer group: XGROUP CREATE telemetry:events analytics_group $ MKSTREAM.
3. Consume batch: XREADGROUP GROUP analytics_group worker_1 COUNT 10 BLOCK 2000 STREAMS telemetry:events >.
4. Acknowledge processed entries: XACK telemetry:events analytics_group message_id.

## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.

import Redis from 'ioredis';

export const redis = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379');

export async function initConsumerGroup(streamKey: string, groupName: string): Promise<void> {
  try {
    // MKSTREAM ensures the stream is created if it does not already exist
    await redis.xgroup('CREATE', streamKey, groupName, '$', 'MKSTREAM');
  } catch (err: any) {
    if (!err.message.includes('BUSYGROUP')) {
      throw err;
    }
  }
}

import { redis, initConsumerGroup } from './streamConsumer';

const STREAM_KEY = 'telemetry:events';
const GROUP_NAME = 'analytics_group';
const CONSUMER_NAME = `worker_${process.pid}`;

export async function startWorker() {
  await initConsumerGroup(STREAM_KEY, GROUP_NAME);

  while (true) {
    try {
      // Read only new messages targeting this group using special ID '>'
      const response = await redis.xreadgroup(
        'GROUP', GROUP_NAME, CONSUMER_NAME,
        'COUNT', 10,
        'BLOCK', 2000,
        'STREAMS', STREAM_KEY,
        '>'
      ) as [string, [string, string[]][]][];

      if (!response || response.length === 0) continue;

      const [_, messages] = response[0];
      for (const [id, fields] of messages) {
        await processRecord(id, fields);
        await redis.xack(STREAM_KEY, GROUP_NAME, id);
      }
    } catch (error) {
      console.error('Consumer error in stream loop:', error);
      await new Promise((resolve) => setTimeout(resolve, 1000));
    }
  }
}

async function processRecord(id: string, rawFields: string[]) {
  // Ingestion parsing logic
}