# Terminal 1: Subscribe to a channel
redis-cli SUBSCRIBE notifications:user:42
# Reading messages... (press Ctrl-C to quit)
# 1) "subscribe"
# 2) "notifications:user:42"
# 3) (integer) 1    ← number of channels subscribed

# Terminal 2: Publish a message
redis-cli PUBLISH notifications:user:42 '{"type":"order_shipped","orderId":"ORD-9871"}'
# → (integer) 1   ← number of subscribers that received the message

# Terminal 1 receives:
# 1) "message"
# 2) "notifications:user:42"
# 3) "{\"type\":\"order_shipped\",\"orderId\":\"ORD-9871\"}"

# Subscribe to all channels matching a glob pattern
redis-cli PSUBSCRIBE "notifications:user:*"
# Matches: notifications:user:42, notifications:user:99, notifications:user:any-value

redis-cli PSUBSCRIBE "cache:*:invalidate"
# Matches: cache:product:invalidate, cache:category:invalidate, etc.

// ✅ XADD — Append an entry to a stream
const entryId = await redis.xadd(
  'orders:events',              // Stream key
  '*',                          // Auto-generate entry ID: {milliseconds}-{sequence}
  'orderId',  'ORD-9871',
  'userId',   '42',
  'type',     'order_placed',
  'amount',   '149.99'
)
// entryId: "1725350400000-0"  ← {millisecond_timestamp}-{sequence_within_ms}

// Bounded stream: auto-trim to 100,000 entries (MAXLEN with ~ for approximate)
await redis.xadd('orders:events', { MAXLEN: ['~', 100000] }, '*',
  'orderId', 'ORD-9872', 'type', 'order_shipped'
)

# Read all entries from the beginning
redis-cli XRANGE orders:events - +
# 1) 1) "1725350400000-0"
#    2) 1) "orderId" 2) "ORD-9871" 3) "userId" 4) "42" ...

# Read the last 10 entries
redis-cli XREVRANGE orders:events + - COUNT 10

# Blocking read: wait for new entries (like BLPOP for Streams)
redis-cli XREAD BLOCK 5000 STREAMS orders:events $
# BLOCK 5000 = block for up to 5000ms waiting for new entries
# $           = only return entries added AFTER this command started

// Step 1: Create a consumer group (run once at startup)
// '$' = start from new entries only (use '0' to process existing entries from the beginning)
await redis.xgroup('CREATE', 'orders:events', 'processors', '$', 'MKSTREAM')
// MKSTREAM: create the stream if it doesn't exist

// Step 2: Read up to 10 pending entries as consumer-1
const entries = await redis.xreadgroup(
  'GROUP', 'processors', 'consumer-1',
  'COUNT', 10,
  'BLOCK', 5000,        // Block up to 5s waiting for new entries
  'STREAMS', 'orders:events', '>'
  // '>' means: give me entries not yet delivered to any consumer in this group
)
// entries: [['orders:events', [['1725350400000-0', ['orderId', 'ORD-9871', ...]]]]]

// Step 3: Process and acknowledge
for (const [streamKey, messages] of entries) {
  for (const [entryId, fields] of messages) {
    await processOrder(fields)      // Your business logic
    await redis.xack('orders:events', 'processors', entryId)
    // Remove from PEL — entry will not be re-delivered
  }
}

# Summary: how many entries are pending in the consumer group?
redis-cli XPENDING orders:events processors - + 10
# 1) 1) "1725350400001-0"    ← entry ID
#    2) "consumer-2"          ← which consumer it was delivered to
#    3) (integer) 45231       ← idle time in ms (45 seconds — consumer likely crashed)
#    4) (integer) 1           ← delivery count (delivered once)

# Full summary
redis-cli XPENDING orders:events processors
# 1) (integer) 1        ← 1 entry pending in this group
# 2) "1725350400001-0"  ← min pending entry ID
# 3) "1725350400001-0"  ← max pending entry ID
# 4) 1) 1) "consumer-2" 2) "1"   ← consumer-2 has 1 pending entry

// XAUTOCLAIM: claim entries from consumer-2 that have been idle > 30 seconds
const [nextCursor, claimedEntries] = await redis.xautoclaim(
  'orders:events',    // Stream key
  'processors',       // Consumer group
  'consumer-3',       // New owner
  30000,              // Minimum idle time in ms
  '0-0',             // Start from the beginning of the PEL
  'COUNT', 10         // Claim at most 10 entries
)
// Process and ack claimed entries — same as normal XREADGROUP processing

const MAX_DELIVERY_ATTEMPTS = 3

async function processWithDLQ(entryId: string, fields: string[], deliveryCount: number) {
  if (deliveryCount > MAX_DELIVERY_ATTEMPTS) {
    // Route to DLQ stream for manual inspection
    await redis.xadd('orders:events:dlq', '*',
      'originalId', entryId,
      'failureReason', 'max_retries_exceeded',
      'deliveryCount', String(deliveryCount),
      ...fields
    )
    // Acknowledge the original to remove from PEL — it's in the DLQ now
    await redis.xack('orders:events', 'processors', entryId)
    return
  }
  await processOrder(fields)
  await redis.xack('orders:events', 'processors', entryId)
}

# MAXLEN — trim to a maximum number of entries
redis-cli XADD orders:events MAXLEN 100000 * orderId ORD-9873 type order_placed
# '~' prefix = approximate trim (Redis may keep slightly more for efficiency)
redis-cli XADD orders:events 'MAXLEN' '~' 100000 '*' orderId ORD-9874 type order_shipped

# MINID — trim entries with IDs older than a given ID (time-based retention)
redis-cli XTRIM orders:events MINID 1725264000000  # Remove entries older than 24 hours ago

# Explicit trim (separate from XADD)
redis-cli XTRIM orders:events MAXLEN ~ 50000

# BLMPOP: blocking pop from the left of one or more lists
redis-cli BLMPOP 5 2 jobs:high jobs:normal LEFT COUNT 5
# Waits up to 5 seconds
# Checks 2 lists: jobs:high first (priority), then jobs:normal
# Pops from LEFT, returns up to 5 elements
# Returns from the first non-empty list

# vs. old BLPOP (single list, single element only)
redis-cli BLPOP jobs:high 5