Streams

# 1) Append events
XADD events * type signup user 42
XADD events * type purchase order 998 amount 99.95
# Auto-generated IDs are <ms>-<seq>; you can pass your own.

# 2) Read by ID range
XRANGE events - +                       # all events
XRANGE events 1717000000000-0 +         # since a timestamp
XREVRANGE events + - COUNT 10           # last 10

# 3) Read as a stream — blocks until new arrivals
XREAD COUNT 10 BLOCK 5000 STREAMS events $
# $ = only NEW events from now
# 0 = from the beginning

# 4) Consumer groups — many workers share the stream + each gets a slice
XGROUP CREATE events workers $ MKSTREAM       # MKSTREAM creates if missing
XREADGROUP GROUP workers worker-1 \
    COUNT 10 BLOCK 5000 \
    STREAMS events >
# > = &ldquo;new messages I haven't seen&rdquo;

# 5) Acknowledge processed events
XACK events workers $message_id

# 6) Inspect pending — what's been delivered but not ACKed
XPENDING events workers
XPENDING events workers - + 10 worker-1

# 7) Claim stale messages — fault-tolerant worker pattern
#    A worker crashed; another claims its in-flight messages.
XCLAIM events workers worker-2 60000 $message_id
XAUTOCLAIM events workers worker-2 60000 0 COUNT 10

# 8) Cap the stream — bounded memory, time-window or count
XADD events MAXLEN ~ 1000000 * type page_view path /home
# ~ = approximate (faster, near 1M)

XTRIM events MAXLEN 1000000
XTRIM events MINID 1717000000000-0      # drop anything older

# 9) Length + status
XLEN events
XINFO STREAM events FULL
XINFO GROUPS events
XINFO CONSUMERS events workers

# 10) Node + ioredis worker loop
const redis = new Redis();
await redis.xgroup('CREATE', 'events', 'workers', '$', 'MKSTREAM').catch(() => {});

while (true) {
    const res = await redis.xreadgroup(
        'GROUP', 'workers', \`worker-${id}\`,
        'COUNT', 10, 'BLOCK', 5000,
        'STREAMS', 'events', '>',
    );
    if (!res) continue;
    for (const [, messages] of res) {
        for (const [id, fields] of messages) {
            try {
                await handle(parseFields(fields));
                await redis.xack('events', 'workers', id);
            } catch (e) { /* will retry after XCLAIM */ }
        }
    }
}

XADD events * type signup user 42
XREAD COUNT 10 STREAMS events 0
XLEN events
XGROUP CREATE events workers '$' MKSTREAM