@beignet/provider-broadcast-redis

Redis Pub/Sub BroadcastPort provider for Beignet applications with multiple request instances or worker processes. It carries ephemeral browser update hints. Missed messages cannot be replayed; browsers must refetch after reconnect.

Install

bun add @beignet/core @beignet/provider-broadcast-redis ioredis zod

Beignet requires Node.js 22.12 or newer. Bun is also supported. This provider uses Node.js networking APIs and is not an Edge adapter. The example below uses Zod for channel schemas.

Configure the provider

import { createRedisBroadcastProvider } from "@beignet/provider-broadcast-redis";

export const providers = [createRedisBroadcastProvider()] as const;

Add this to your existing provider registry. Declare broadcast: BroadcastPort in AppPorts, importing the type from @beignet/core/broadcasting/server, and defer broadcast in infra/port-wiring.ts. The CLI can apply that wiring:

bun beignet provider add broadcast-redis
bun install

| Environment variable | Meaning | Default | | --- | --- | --- | | REDIS_BROADCAST_URL | Redis connection URL | Required | | REDIS_BROADCAST_PREFIX | Channel prefix unique to your app/environment | beignet:broadcast: | | REDIS_BROADCAST_CONNECT_TIMEOUT_MS | Connection and command deadline, 1–30,000 ms | 5000 |

Options { url, prefix, connectTimeoutMs, name } override the corresponding environment settings; name changes the lifecycle provider name. Use a rediss:// URL when your Redis deployment requires TLS. Every publisher and subscriber must use the same prefix. Redis Pub/Sub database numbers do not isolate channels; use separate prefixes and appropriate Redis access controls.

Direct use

import { defineChannel } from "@beignet/core/broadcasting";
import { createRedisBroadcast } from "@beignet/provider-broadcast-redis";
import { z } from "zod";

const changes = defineChannel("issues.changes", {
  params: z.object({ workspaceId: z.string() }),
  events: { changed: z.object({ issueId: z.string() }) },
});
const broadcast = createRedisBroadcast({
  url: "redis://localhost:6379",
  prefix: "my-app:development:broadcast:",
});
await broadcast.start();
const received = Promise.withResolvers<void>();
const subscription = broadcast.subscribe(changes, {
  params: { workspaceId: "workspace-1" },
  onEvent: event => {
    console.log(event.data.issueId);
    received.resolve();
  },
  onDisconnect: () => console.log("Reconnect and refetch"),
});
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
  await subscription.ready;
  await broadcast.publish(changes, {
    params: { workspaceId: "workspace-1" },
    event: "changed",
    data: { issueId: "issue-1" },
  });
  await Promise.race([
    received.promise,
    new Promise<never>((_, reject) => {
      timeout = setTimeout(() => reject(new Error("No hint received")), 5_000);
    }),
  ]);
} finally {
  clearTimeout(timeout);
  try {
    await subscription.unsubscribe();
  } finally {
    await broadcast.close();
  }
}

Start a local Redis server before running this example. It waits up to five seconds to print issue-1 before shutting down; a successful publish() alone does not confirm that the listener ran. An application keeps its subscription until the owning component or process shuts down, and handles gaps by reconnecting and refetching authoritative state.

RedisBroadcast extends BroadcastPort with start() and close(). createRedisBroadcast requires url and accepts prefix, connectTimeoutMs, and instrumentation. The lifecycle provider starts and closes it for you.

Each instance shares a dedicated publisher/subscriber connection pair and reference-counts subscriptions to the same channel. Offline command queues, automatic subscription replay, and resending unfinished commands are disabled. Initial connection setup is bounded; failed provider startup closes both clients. Runtime connections may reconnect, but existing subscriptions report continuity loss and require fresh subscriptions. createBroadcastRoute closes affected SSE streams so browser clients reconnect and reconcile.

Publication resolves when Redis accepts the message, even with zero subscribers. It does not confirm browser delivery. Retries can duplicate hints. Schemas are validated before publishing and on receipt. Topic keys hash canonical channel identity; prefixes and hashes are not authorization boundaries.

Instrumentation uses the broadcast watcher for publications, failures, subscription lifecycle, malformed envelopes, and connection gaps. Records omit payloads, parameters, connection URLs, and origin IDs.

For required publication after a database mutation, record a publication job in the same transaction through an outbox and run it after commit. The outbox makes the publication attempt retryable; it does not make Redis subscribers durable.

Serverless deployment

Use a host that supports streaming HTTP responses and outbound Redis sockets. Configure SSE renewal below the host's request deadline with setup and cleanup headroom. Keep providers shared within an instance; do not create Redis clients for each channel or start a polling worker in a route module. Measure Redis connection counts and request memory/cost at your expected concurrency.

See the broadcasting guide for authorized channels, browser recovery, and deployment limits. For provider development, REDIS_TEST_URL=redis://localhost:6379 bun run test:live exercises independent connections and recovery after a real socket interruption.