@beignet/provider-event-bus-redis

Runtime: Beignet requires Node.js 22.12 or newer. Bun is optional.

Redis Pub/Sub EventBusPort provider for Beignet applications.

Use it when a multi-process app needs domain events published by one process to reach listeners registered in another process. Delivery is best-effort: subscribers must be online, Redis does not persist messages for later replay, and the provider does not acknowledge, retry, or dead-letter failed handlers. Use Beignet outbox and jobs for durable side effects that need retry, idempotency, or recovery after downtime.

createRedisEventBusProvider(...) returns the stable RedisEventBusProvider type. RedisEventBusConfig describes its validated config; the Zod schema remains internal.

Install

bun add @beignet/provider-event-bus-redis @beignet/core ioredis
Copy

You can also wire it into an existing app with the CLI:

beignet provider add event-bus-redis
Copy

Setup

import { createNextServer, createNextServerLoader } from "@beignet/next";
import { createRedisEventBusProvider } from "@beignet/provider-event-bus-redis";
import { initialPorts } from "@/infra/port-wiring";
import { appContext } from "@/server/context";
import { routes } from "@/server/routes";

// Set environment variables:
// REDIS_EVENT_BUS_URL=redis://localhost:6379
// REDIS_EVENT_BUS_DB=0 (optional)

export const getServer = createNextServerLoader(() =>
  createNextServer({
    ports: initialPorts,
    providers: [createRedisEventBusProvider()],
    context: appContext,
    routes,
  }),
);
Copy

The provider contributes ctx.ports.eventBus, the standard Beignet event bus port, and ctx.ports.redisEventBus as a Redis-specific escape hatch.

Configuration

The provider reads environment variables with the REDIS_EVENT_BUS_ prefix:

Factory options

Use createRedisEventBusProvider(options) when the app should own connection settings. Defined options override matching REDIS_EVENT_BUS_* environment variables:

import { createRedisEventBusProvider } from "@beignet/provider-event-bus-redis";

const provider = createRedisEventBusProvider({
  channelPrefix: "my-app:events:",
  connectTimeoutMs: 2000,
  cleanupTimeoutMs: 5000,
  maxRetriesPerRequest: 1,
  db: 1,
  retryStrategy: (times) => Math.min(times * 100, 1000),
});
Copy

Calling createRedisEventBusProvider() with no options uses the env-backed defaults above.

Connection behavior

During startup the provider creates separate publisher and subscriber clients, connects both eagerly, and fails boot with a redacted connection error when Redis is unreachable. The default initial connection retry strategy stops after REDIS_EVENT_BUS_CONNECT_MAX_ATTEMPTS; after a successful connection, ioredis reconnects with capped exponential backoff and resubscribes to the channels owned by that subscriber client. Events published while the subscriber is disconnected are still lost because Redis Pub/Sub has no replay. If an initial SUBSCRIBE command fails after ioredis exhausts its command retries, the adapter retries that desired subscription with capped backoff for as long as at least one local handler remains registered.

Deployment topology

The environment-backed provider targets a standalone Redis URL or a managed Redis service that exposes one Redis-compatible endpoint. It opens two TCP connections per app process: one publisher and one subscriber. Include both in the process connection budget.

Every long-lived process that should receive events must install the provider and register its listeners. Ephemeral request runtimes can publish events, but they are not reliable subscriber hosts because Redis only delivers while the connection is online. Run subscriptions in a long-lived web or worker process.

Sentinel and Redis Cluster require constructor options that cannot be expressed by REDIS_EVENT_BUS_URL. Apps using those topologies should create their own ioredis clients and pass them to createRedisEventBus(...). Beignet's live suite does not currently prove Sentinel or Cluster failover semantics, so test subscription recovery against the exact managed topology before relying on it. HTTP-only Redis APIs are not compatible because Pub/Sub requires a persistent Redis connection.

Direct adapter

Use createRedisEventBus(...) when your app owns Redis clients:

import Redis from "ioredis";
import { defineEvent } from "@beignet/core/events";
import { createRedisEventBus } from "@beignet/provider-event-bus-redis";
import { z } from "zod";

const UserRegistered = defineEvent("user.registered", {
  payload: z.object({ userId: z.string() }),
});

function handleUserRegistered(payload: { userId: string }) {
  console.log(`Registered ${payload.userId}`);
}

const publisher = new Redis(process.env.REDIS_EVENT_BUS_URL);
const subscriber = new Redis(process.env.REDIS_EVENT_BUS_URL);
const eventBus = createRedisEventBus({
  publisher,
  subscriber,
  channelPrefix: "my-app:events:",
});

const subscription = eventBus.subscribe(UserRegistered, handleUserRegistered);
await subscription.ready;

// During app shutdown:
try {
  await eventBus.stop();
  await Promise.all([publisher.quit(), subscriber.quit()]);
} catch (error) {
  subscriber.disconnect();
  publisher.disconnect();
  throw error;
}
Copy

Register listeners during startup before handling traffic and await each subscription's ready promise. It resolves after Redis acknowledges the channel subscription. Failed subscription commands retry with capped backoff while a local handler remains; callers can own a deadline, and registerListeners(...) applies one 10-second deadline to a complete registry by default. Readiness proves initial registration only and does not change Pub/Sub's online-only delivery semantics.

Await unsubscribe() when removing an individual handler. During adapter shutdown, stop() closes all remaining subscriptions, detaches local delivery, cancels pending retries, and awaits channel cleanup. Cleanup errors reject and are instrumented; stop() retries channels left dirty by an earlier unsubscribe before reporting any remaining failure. To keep lifecycle rollback bounded, stop() rejects after cleanupTimeoutMs (10 seconds by default) even if an ioredis command never settles. The adapter does not close caller-owned Redis clients, so force-disconnect them when cleanup rejects. Stopping is terminal: later subscription attempts reject instead of creating handlers that cannot receive messages.

Delivery semantics

• publish(...) publishes one Redis Pub/Sub message on <channelPrefix><event.name>.
• Only subscribers that are online and currently subscribed receive the event.
• A subscription's ready promise resolves only after the initial Redis subscription acknowledgement. Cancelling before acknowledgement rejects that promise and prevents future local delivery.
• ioredis resubscribes after a transient reconnect, but messages published during the disconnected interval are not recovered.
• Payloads are JSON encoded and delivered without persisting the message. Direct publication, Beignet's use-case event helpers, and publishEvent(...) reject non-JSON output and transforms whose canonical JSON changes when validated again. registerListeners(...) repeats the stability check on decoded JSON before handler execution.
• When a tracing port is available, the envelope includes Beignet's versioned trace carrier and subscribers receive it as publish metadata. Registered listeners continue the producer trace. Legacy messages without a carrier and messages with malformed trace metadata still deliver their payload.
• Handler errors are reported through onHandlerError when provided and are also recorded as provider instrumentation events; they do not reject the original publisher.
• Invalid messages and subscription errors are reported through onSubscriberError when provided and recorded as eventBus.message.failed or eventBus.subscription.failed provider instrumentation events. Failed subscriptions retry with capped backoff while their local handlers remain registered; removing the last handler cancels a pending retry.
• Failed publish(...) calls reject with the Redis error and record an eventBus.publish.failed instrumentation event. The provider does not retry the publish after the configured ioredis command retries are exhausted.
• When the eventBus instrumentation watcher is enabled, publisher and subscriber connection errors from env-backed clients are recorded as eventBus.connection.failed. When instrumentation is absent or disabled, Beignet leaves ioredis's error fallback intact so connection failures remain visible.
• The provider does not retry handlers, store attempts, replay missed events, or dead-letter failures.

When side effects must survive process restarts or dependency outages, record the domain event through the Unit of Work outbox and drain it through jobs or outbox-backed listeners.

Escape hatch and health

ctx.ports.redisEventBus exposes:

const health = await ctx.ports.redisEventBus.checkHealth();

ctx.ports.redisEventBus.publisher;
ctx.ports.redisEventBus.subscriber;
ctx.ports.redisEventBus.channelPrefix;
Copy

checkHealth() sends cheap PING commands to both Redis clients and returns metadata suitable for app-owned readiness endpoints. It proves both connections can reach Redis; it does not prove that every expected listener has registered its channel subscription.

Instrumentation

The provider records successful publish events under the eventBus watcher. Received messages, failed publishes, connection failures, subscription failures, and handler failures are recorded as custom provider events. Payloads are not recorded.

Testing

Use @beignet/provider-event-bus-memory for deterministic unit tests and single-process local tests. Use this Redis provider in integration tests when the behavior under test specifically depends on cross-process delivery or Redis connection behavior.

This package includes opt-in live Redis integration tests for cross-instance delivery, reconnect and automatic resubscription, expected loss while offline, active-subscription cleanup during shutdown, readiness, instrumentation, and failure callbacks. The default bun run test command runs the hermetic unit suite only so the live tests cannot share process state with the mocked ioredis tests. Set REDIS_EVENT_BUS_TEST_URL or the shared REDIS_TEST_URL before running the dedicated live test command:

REDIS_EVENT_BUS_TEST_URL=redis://localhost:6379 bun run test:live
Copy