Redis Client

Redis Client is a lightweight, type-safe wrapper for ioredis. It solves the problem of scaling Redis infrastructure by automatically routing Write operations to a Primary node and Read operations to Replica nodes.

📑 Table of Contents

• Features
• Prerequisites
• Installation
• Usage
• API Reference
• Configuration
• Architecture
• Running Tests
• Contributing
• License

✨ Features

• Automatic Traffic Splitting: Writes (set, hset, del, expire) are sent to the Primary; Reads (get, hget, hgetall, scan, keys) are sent to Replicas.
• Type Safety: Built with TypeScript, providing full IntelliSense and type checking out of the box.
• Performance: Offloads expensive read operations (like SCAN) from your Master node to prevent blocking critical write traffic.
• Flexible Access: Exposes the underlying ioredis instances if you need to run custom or unsupported commands.

🔧 Prerequisites

Before you begin, ensure you have met the following requirements:

• Node.js (v20 or higher — required by ioredis 6.x).
• Redis: A Redis setup with at least one Master and one Replica (e.g., AWS ElastiCache Cluster Mode Disabled).

🚀 Installation

Install the package (ioredis is bundled as a dependency — no separate install needed):

npm install @heady/redis-client

💡 Usage

import { RedisClient } from '@heady/redis-client';

// 1. Initialize with separate endpoints
const redis = new RedisClient({
  primary: {
    host: 'primary-node.redis.aws.internal',
    port: 6379,
  },
  replica: {
    host: 'replica-node.redis.aws.internal',
    port: 6379,
  }
});

async function main() {
  // ✍️ WRITE: Automatically routed to Primary
  await redis.set('user:101', 'John Doe', 3600); 
  console.log('User saved to Primary node');

  // 📖 READ: Automatically routed to Replica
  const user = await redis.get('user:101');
  console.log('User read from Replica node:', user);

  // 🧹 CLEANUP
  await redis.disconnect();
}

main();

Streams (consumer groups)

The client has no stream methods, so run stream commands on the primary with getWriteClient(). XREADGROUP, XAUTOCLAIM and XACK all change consumer-group state, so they must not go to a replica. ioredis returns stream replies as unknown; cast them to the exported types and convert fields with fieldsToObject.

import {
  RedisClient,
  XReadGroupReply,
  XAutoClaimReply,
  fieldsToObject,
} from '@heady/redis-client';

const STREAM = 'orders';
const GROUP = 'order-workers';
const CONSUMER = 'worker-1';

// BLOCK holds the connection open, so give the consumer its own connection
const stream = redis.getWriteClient().duplicate();

async function handle(id: string, fields: string[] | null) {
  if (fields === null) return; // entry was deleted while pending (Redis 6.2)
  const order = fieldsToObject(fields); // { orderId: '42', status: 'new' }
  // ...process order...
  await stream.xack(STREAM, GROUP, id);
}

// 1. Reclaim messages left un-ACKed by crashed consumers
async function reclaimPending() {
  let startId = '0-0';
  do {
    const [nextStartId, entries] = (await stream.xautoclaim(
      STREAM, GROUP, CONSUMER,
      60_000,  // only claim messages idle for at least 60s
      startId, // cursor
      'COUNT', 10,
    )) as XAutoClaimReply;
    for (const [id, fields] of entries) await handle(id, fields);
    startId = nextStartId;
  } while (startId !== '0-0'); // "0-0" = whole pending list scanned
}

// 2. Read new messages
async function readNew() {
  const reply = (await stream.xreadgroup(
    'GROUP', GROUP, CONSUMER,
    'COUNT', 10,
    'BLOCK', 5000,
    'STREAMS', STREAM, '>',
  )) as XReadGroupReply;
  if (!reply) return; // timed out, nothing new
  for (const [, entries] of reply) {
    for (const [id, fields] of entries) await handle(id, fields);
  }
}

📚 API Reference

All methods are async and return Promises. Routing is automatic.

Write operations → Primary

| Method | Signature | Description | |:--|:--|:--| | set | set(key, value, ttlSeconds?) | Set a string/number value; optional TTL in seconds (EX). | | hset | hset(key, object) | Set multiple hash fields from an object. Returns count of new fields. | | del | del(key) | Delete a key. Returns number removed. | | expire | expire(key, seconds) | Set a key's TTL. |

Read operations → Replica

| Method | Signature | Description | |:--|:--|:--| | get | get(key) | Get a string value, or null. | | hget | hget(key, field) | Get a single hash field, or null. | | hgetall | hgetall(key) | Get all fields of a hash as an object. | | scan | scan(cursor, matchPattern, count?) | Cursor-based scan (MATCH/COUNT, default count 100). Returns [nextCursor, keys[]]. | | keys | keys(pattern) | List keys matching a pattern. Use scan in production — keys blocks the node. |

Utilities

| Method | Signature | Description | |:--|:--|:--| | getWriteClient | getWriteClient() | Underlying primary ioredis instance for custom commands. | | getReadClient | getReadClient() | Underlying replica ioredis instance for custom commands. | | disconnect | disconnect() | Gracefully quit() both connections. |

Stream helpers

| Export | Kind | Description | |:--|:--|:--| | fieldsToObject | fieldsToObject(fields: string[]) | Turns a flat stream field list ['k1','v1','k2','v2'] into { k1: 'v1', k2: 'v2' }. | | StreamEntry | type | [id, fields \| null]. fields is null for an entry deleted while pending. | | XReadGroupReply | type | XREADGROUP reply: [stream, StreamEntry[]][], or null on BLOCK timeout. | | XAutoClaimReply | type | XAUTOCLAIM reply: [nextStartId, StreamEntry[], deletedIds?]. deletedIds requires Redis 7+. |

⚙️ Configuration

The RedisClient constructor accepts a configuration object with two keys: primary and replica. Both accept standard ioredis options.

| Property | Type | Description | |:-----|:--------:|------:| | primary | RedisOptions | Configuration for the Master (Write) node. | | replica | RedisOptions | Configuration for the Read Replica

Example Config Object:

{
  primary: { host: '127.0.0.1', port: 6379, password: 'auth' },
  replica: { host: '127.0.0.1', port: 6380, password: 'auth' }
}

🏗 Architecture

This library implements the Read/Write Splitting pattern:

1. Command Interception: The client checks the method being called (e.g., set vs get).
2. Routing: * Mutating commands → Primary Connection
   • Read-only commands → Replica Connection
3. Result: This ensures high availability and prevents heavy read queries from slowing down data ingestion.

🧪 Running Tests

This project uses Jest with ts-jest for unit testing. The tests mock ioredis to ensure traffic is routed to the correct connection without needing a real Redis instance.

To run the test suite:

npm test

Expected Output:

 PASS  test/client.test.ts
  RedisClient
    ✓ should create two separate Redis connections
    ✓ set() should call primary node
    ✓ get() should call replica node

📝 Changelog

See CHANGELOG.md for release history.

📜 License

This project is licensed under the ISC License (see the license field in package.json).