AbsoluteJS

@absolutejs/sync-bus-redis

@absolutejs/sync-bus-redisv0.1.2betaData & Sync

Redis pub/sub ClusterBus for @absolutejs/sync — cross-instance fan-out via PUBLISH/SUBSCRIBE. Sibling to @absolutejs/sync-bus-pg; faster fanout, better geo-replication story. Works with any Redis client (ioredis, node-redis, etc.) via a narrow tag-template interface.

#Installation

BASH
bun add @absolutejs/sync-bus-redis

#Capabilities

Overview

Redis pub/sub ClusterBus for @absolutejs/sync. Sibling to @absolutejs/sync-bus-pg — same ClusterBus contract, different transport.

Docs: absolutejs.com/documentation/cluster-bus-overview#redis-adapter

When to use Redis vs Postgres

Concern — sync-bus-redis — sync-bus-pg

Payload size — No cap — JSON through — 8KB NOTIFY cap (spill-table fallback)

Fan-out latency at 10+ subscribers — In-memory, low — WAL-replicated, higher

Show 4 more

Geo-replication — Native (Redis Cluster, ElastiCache, Memorystore, Upstash) — Heavy ops (PG logical replication)

Delivery semantics — At-most-once (no message retention) — At-most-once (NOTIFY) + spill rows for oversized

Already in your stack? — If yes → win-win — If yes → win-win

The headline tradeoff. Redis is in-memory pub/sub — a subscriber that's disconnected when a message fires misses it. For cross-instance resume past shard reboot, pair with engine.exportChangeLog() / importChangeLog() (sync 1.19.0+) regardless of which bus you use.

API

Returns ClusterBus & { metrics() }. Pass to engine.connectCluster(bus).

bus.metrics()

Outcomes

What you can build

Build on the supported package contract

Use @absolutejs/sync-bus-redis through its supported public entry points.

Hardening checklist

Production guidance

Make every external boundary explicitPin the deployed @absolutejs/sync-bus-redis version, replace example or memory-backed dependencies with durable implementations, bound external calls, protect credentials, and emit enough evidence to retry or recover safely.

Follow in order

Troubleshooting path

1
Trace from the first failed boundary
Reproduce the smallest canonical @absolutejs/sync-bus-redis example, confirm the supported entry point and version in the API explorer, then inspect the first boundary that did not produce its documented result.

#Usage with ioredis

Partial snippet

Working example for Usage with ioredis.

TS
import { Redis } from 'ioredis';
import { createSyncEngine } from '@absolutejs/sync/engine';
import { createRedisClusterBus } from '@absolutejs/sync-bus-redis';

const publisher = new Redis(process.env.REDIS_URL!);
const subscriberClient = new Redis(process.env.REDIS_URL!);

// ioredis: a subscribed connection can't issue other commands.
// Bridge its EventEmitter API into our (channel, listener) shape:
const subscriber = {
  subscribe: async (channel: string, listener: (msg: string) => void) => {
    await subscriberClient.subscribe(channel);
    const handler = (chan: string, msg: string) => {
      if (chan === channel) listener(msg);
    };
    subscriberClient.on('message', handler);
    return async () => {
      subscriberClient.off('message', handler);
      await subscriberClient.unsubscribe(channel);
    };
  },
};

const bus = createRedisClusterBus({ publisher, subscriber });
const engine = createSyncEngine({ instanceId: 'shard-A' });
await engine.connectCluster(bus);

#Usage with node-redis v4+

Partial snippet

Working example for Usage with node-redis v4+.

TS
import { createClient } from 'redis';
import { createRedisClusterBus } from '@absolutejs/sync-bus-redis';

const publisher = createClient({ url: process.env.REDIS_URL });
const subscriberClient = publisher.duplicate();
await Promise.all([publisher.connect(), subscriberClient.connect()]);

// node-redis: subscribe takes a callback directly.
const subscriber = {
  subscribe: async (channel: string, listener: (msg: string) => void) => {
    await subscriberClient.subscribe(channel, listener);
    return async () => {
      await subscriberClient.unsubscribe(channel);
    };
  },
};

const bus = createRedisClusterBus({ publisher, subscriber });

#API

Partial snippet

Working example for API.

TS
createRedisClusterBus({
  publisher,         // RedisPublisher — publish(channel, message)
  subscriber,        // RedisSubscriber — subscribe(channel, listener) → unsubscribe fn
  channel?,          // default 'absolutejs_sync_cluster'
  onError?,          // logger for parse / delivery failures
});

#bus.metrics()

Partial snippet

Working example for bus.metrics().

TS
{
  published: number;             // PUBLISH calls
  received: number;              // messages parsed + delivered to onMessage
  publishErrors: number;         // publisher.publish() rejections
  subscribeErrors: number;       // JSON.parse failures on incoming messages
  totalSubscribersReached: number; // sum of counts Redis returned from PUBLISH
                                 // — a drop to 0 when you expect peers signals
                                 // subscriber disconnects (partition, restart)
}

#Public entry points

Supported entry points declared by this package manifest.

Package entry point declared in package.json.

@absolutejs/sync-bus-redis@absolutejs/sync-bus-redis/manifest@absolutejs/sync-bus-redis/manifest.json

#Package commands

Scripts declared by this package manifest.

bun run buildrm -rf dist && bun build src/index.ts src/manifest.ts --outdir dist --sourcemap --target=bun --external @absolutejs/sync && tsc --project tsconfig.build.json && absolute-manifest emit
bun run formatprettier --write "./**/*.{ts,json,md}"
bun run testbun test
bun run typechecktsc --noEmit