Broadcasting

Use broadcasts to refresh issue lists, unread counts, job progress, and other browser state when something changes. Define a typed channel, authorize each subscription, and publish small hints through ctx.ports.broadcast. Browsers receive those hints over one multiplexed Server-Sent Events (SSE) connection.

The database remains authoritative. Broadcasts are ephemeral: disconnected browsers miss events, duplicates are possible, and publication does not prove that a browser received anything. Refetch the channel's state after initial readiness and every reconnection. This capability does not provide durable replay, presence, offline writes, or collaborative conflict resolution.

Add broadcasting

bun beignet make broadcast issues.changes
bun install

The generator adds a browser-safe channel, a fail-closed authorization use case and binding, lib/broadcasting.ts, server/broadcasts.ts, a streaming endpoint, and a memory provider when the broadcast port is missing. Replace the authorization denial with your application's current access checks. Run bun install after generation to install the added provider dependencies. It reuses auth.required() when your app already has the conventional lib/route-auth.ts authentication helper. Otherwise the binding must assert identity itself. It does not attach publications to unrelated mutations.

For multiple processes, select Redis before generating the first channel:

bun beignet provider add broadcast-redis
bun beignet make broadcast issues.changes
bun install

Set REDIS_BROADCAST_URL and an application/environment-specific REDIS_BROADCAST_PREFIX. If memory is already installed, replace its provider registration with Redis. Every publisher and subscriber must use the same Redis deployment and prefix.

Define the browser contract

Keep channel definitions in features/issues/channels.ts. Import this module from both the browser and server; it must not import authorization or runtime composition modules.

import { defineChannel } from "@beignet/core/broadcasting";
import { z } from "zod";

export const issuesChanges = defineChannel("issues.changes", {
  params: z.object({ workspaceId: z.string().min(1) }),
  events: {
    changed: z.object({ issueId: z.string(), key: z.string() }),
  },
});

Parameters must parse to a flat record of strings. Parameter order does not change channel identity. Each event has its own schema and a discriminated payload type. Schemas must produce canonical JSON: plain objects, arrays, strings, finite numbers, booleans, and null. Use timestamp strings instead of Date. Transforms must be deterministic, idempotent, and free of side effects; validation runs at producing and receiving boundaries. Parsed values are validated again after JSON normalization. Invalid or unstable values reject publication with BroadcastValidationError.

Only expose data that every authorized subscriber may read. Prefer identifiers and query invalidation over broadcasting entire database records.

Authorize every subscription

Bind the app context once in lib/broadcasting.ts:

import "@beignet/core/server-only";
import { createBroadcasting } from "@beignet/core/broadcasting/server";
import type { AppContext } from "@/app-context";

export const { defineChannelBinding, defineChannelRegistry } =
  createBroadcasting<AppContext>();

Keep authorization in features/issues/broadcasts.ts. The following binding delegates to the app-owned use case generated by make broadcast:

import { defineChannelBinding } from "@/lib/broadcasting";
import { issuesChanges } from "./channels";
import { authorizeIssuesChangesUseCase } from "./use-cases/authorize-changes-broadcast";

export const issuesChangesBroadcast = defineChannelBinding(issuesChanges, {
  authorize: async ({ ctx, params }) => {
    await authorizeIssuesChangesUseCase.run({ ctx, input: params });
  },
});

That use case must assert the authenticated user, resolve current workspace membership, and enforce any resource visibility policy. A client-supplied workspace ID is a lookup input, not proof of access. Return normally to allow the subscription; throw a catalog error to deny it. Public channels also need an explicit binding with an intentional authorize callback.

Register bindings in server/broadcasts.ts:

import "@beignet/core/server-only";
import { defineChannelRegistry } from "@/lib/broadcasting";
import { issuesChangesBroadcast } from "@/features/issues/broadcasts";

export const channels = defineChannelRegistry([issuesChangesBroadcast]);

Duplicate channel names are rejected. Authorization runs on every connection, including scheduled renewals. Access can remain active until the current stream ends, at most 60 seconds after it opens. Use a shorter lifetime when your revocation requirements need a smaller window.

Expose the stream

In a Next.js app, reuse your existing authentication helper. If the app uses Beignet's session context and has no helper yet, create lib/route-auth.ts:

import { createAuthHooks } from "@beignet/core/server";
import type { AppContext } from "@/app-context";

export const auth = createAuthHooks<AppContext>()({
  resolve: ({ ctx }) => ctx.auth,
});

Then use a thin app/api/broadcasts/route.ts adapter:

import { createBroadcastRoute } from "@beignet/next";
import { auth } from "@/lib/route-auth";
import { getServer } from "@/server";
import { channels } from "@/server/broadcasts";

export const runtime = "nodejs";
export const maxDuration = 120;
export const { GET } = createBroadcastRoute({
  server: getServer,
  channels,
  hooks: [auth.required()],
  maxLifetimeMs: 60_000,
});

The endpoint uses the server's context, hooks, metadata, and HTTP error pipeline. Context fields added by route hooks are available to channel authorization. Set metadata when your app hooks require route metadata.

For a Fetch runtime, import createBroadcastRoute from @beignet/web and mount its GET handler at /api/broadcasts in your host. The web generator exports createAppBroadcastRoute(server) from server/broadcast-route.ts; pass your assembled Fetch server to that factory and mount its returned GET. beignet.config.* can change paths.broadcastRoute, paths.broadcasts, and paths.broadcastingBuilder; the web profile defaults to server/broadcast-route.ts. The endpoint is inspected separately from HTTP contracts and is not added to OpenAPI as a replayable event API.

Publish after the authoritative change

Declare broadcast: BroadcastPort in AppPorts, importing the type from @beignet/core/broadcasting/server. Wire the selected provider in server/providers.ts and defer the port in infra/port-wiring.ts.

await ctx.ports.broadcast.publish(issuesChanges, {
  params: { workspaceId: issue.workspaceId },
  event: "changed",
  data: { issueId: issue.id, key: issue.key },
});

publish resolves when the provider accepts the message, even with zero subscribers. Publishing before a database commit can make a browser refetch old state. For required or retryable publication, record an ordinary job in the same transaction as the mutation through tx.jobs.dispatch(PublishIssueChangeJob, payload). Configure tx.jobs with createOutboxJobDispatcher(transactionOutbox) and register the job with your outbox drain and job execution entrypoint. Its handler publishes the hint. See Outbox for transaction wiring and draining.

For optional hints, you can publish after the write and handle a publication failure without changing the successful mutation result. Await the attempt; do not leave required work running after a serverless request returns.

Subscribe and reconcile

Create one client per browser application session and share it between features. Construct it in browser lifecycle code, then close it when the user or workspace changes. It keeps connection identifiers in memory.

import { createBroadcastClient } from "@beignet/core/broadcasting/client";
import { issuesChanges } from "@/features/issues/channels";

const broadcasts = createBroadcastClient({ url: "/api/broadcasts" });
const subscription = broadcasts.subscribe(issuesChanges, {
  params: { workspaceId },
  onEvent: () => refetchIssues(),
  onSync: () => refetchIssues(),
  onStatusChange: status => showConnectionStatus(status),
  onError: error => reportConnectionError(error),
});

// Component cleanup:
subscription.unsubscribe();
// Session/workspace cleanup:
broadcasts.close();

Here workspaceId, refetchIssues, showConnectionStatus, and reportConnectionError are application values and callbacks. onSync must refetch all state covered by the channel, since events may have been missed while disconnected. Callback failures are reported through onError without breaking other subscribers. Unsubscribing suppresses later callbacks but cannot cancel application work that a callback already started.

The client uses streaming Fetch, same-origin credentials by default, and rejects redirects. Set headers to an async function when credentials need refreshing for each request. Set credentials for your cross-origin cookie policy and configure the matching server CORS policy. Never put credentials in the URL. Identical channel subscriptions share one backend subscription until the last observer unsubscribes.

React Query

Use createBroadcastQuerySubscription from @beignet/react-query to map hints to ordinary TanStack filters. This example assumes existing listIssues and getIssue HTTP contracts in the same issues namespace, with getIssue using a key path parameter. rq is your app's createReactQuery(...) adapter, and queryClient is the TanStack client used by the active UI:

import { createBroadcastQuerySubscription } from "@beignet/react-query";
import { rq } from "@/client";
import { listIssues, getIssue } from "@/features/issues/contracts";
import { issuesChanges } from "@/features/issues/channels";

const subscription = createBroadcastQuerySubscription({
  client: broadcasts,
  channel: issuesChanges,
  params: { workspaceId },
  queryClient,
  invalidates: event => [
    rq(listIssues).contractFilter(),
    rq(getIssue).filter({ path: { key: event.data.key } }),
  ],
  reconciles: [rq(listIssues).namespaceFilter()],
});

Hints mark matching queries stale and coalesce refetches. If a hint arrives during a fetch, one later fetch reconciles it. Active queries refetch; inactive queries stay stale even when an earlier fetch completes after the hint. TanStack's disabled/static query behavior still applies. Teardown removes the subscription and cache listener.

Keep optimistic mutation state separate from remote query results while the mutation is pending. After settlement, reconcile using the returned version and invalidate affected queries. Broadcasting does not resolve conflicting edits or patch application caches automatically.

Exclude the initiating browser

When a mutation already updates its browser's cache, optionally exclude that logical client from its hint. Merge broadcasts.getRequestHeaders() into the typed HTTP client's dynamic headers callback. At the server boundary, capture the header with an authenticated scope:

import { resolveBroadcastOrigin } from "@beignet/core/broadcasting/server";

const broadcastOrigin = resolveBroadcastOrigin({
  headers: request.headers,
  principalId: authenticatedUser.id,
  tenantId: authenticatedWorkspace.id,
  namespace: "my-app",
});

Use the same resolver in createBroadcastRoute({ resolveOrigin }), or return a previously resolved ctx.broadcastOrigin. Pass the captured object as excludeOrigin when publishing. For a job, serialize it as broadcastOrigin in the payload and forward it from the worker; never derive the origin from the worker's service identity. broadcastOriginSchema validates a trusted, previously captured origin at a Standard Schema boundary.

The X-Beignet-Broadcast-Client header contains a random in-memory client ID. It grants no access. Missing or malformed values disable exclusion, as does an anonymous principal. Matching includes the server-owned namespace, tenant, and principal, so another user's client ID cannot suppress their updates. The browser ID survives stream renewals; create a new client when identity changes. Other tabs continue receiving hints. Default redaction hides the header, broadcastOrigin, and excludeOrigin fields.

Notifications and inbox updates

defineBroadcastNotificationChannel({ channel, render }) from @beignet/core/notifications adapts a notification into a typed publication. Return undefined from render to skip delivery. Existing notification preferences and independent channel retry behavior apply. A sent result means provider acceptance, not an online recipient or a stored inbox row.

For a persistent inbox, generate the ordered application recipe:

bun beignet make inbox --broadcast
bun install

make inbox also enables this recipe when the app already declares a broadcast port. When adding broadcasting to an existing generated inbox, rerun make inbox; customized write paths require an explicit application edit. The recipe authorizes the recipient, maps the inbox namespace for list and count reconciliation, and commits each inbox write with an ordinary publication job in the same database transaction. Mount the generated client helper with the app's shared client and authenticated user ID.

Configure a reliable outbox drain. Retrying the publication job only publishes the hint; it never repeats the inbox insert. Original notification delivery retries still need application-owned deduplication where required. Separate inbox and broadcast notification channels run independently and cannot guarantee this ordering.

Limits and recovery

Unknown authorization failures return a sanitized retryable status, without exception details. Malformed protocol messages block the affected connection's subscriptions. Last-Event-ID is rejected: this protocol has no replay cursor. Connection status is connecting, connected, reconnecting, blocked, or closed; use per-subscription status when some channels are blocked. connected means the subscription is ready, not that its query refresh has finished. Server queue overflow closes the affected stream so the browser reconnects and reconciles.

Deployment and operations

Memory broadcasting works only inside one process. Use Redis Pub/Sub when requests, workers, or replicas publish from different processes. Redis uses a shared publisher/subscriber pair per provider instance, bounded connection setup, and asynchronous subscription cleanup. A reconnect cannot recover missed Pub/Sub messages. Use separate prefixes per app and environment, restricted Redis credentials, and TLS where required by the deployment.

Serverless hosts must support streaming responses and outbound connections to Redis. Keep the stream lifetime below the host's request duration, with headroom for context creation, authorization, and cleanup. The Node.js example uses a 60-second stream and a 120-second host duration. Platform limits and plan availability still apply; see Vercel function duration. The Redis provider uses Node.js/ioredis APIs and is not an Edge adapter.

For Bun, configure an idle timeout above the 25-second heartbeat interval; the Vite/Bun example sets idleTimeout: 60 in Bun.serve(...). Bun's default 10-second idle timeout otherwise closes quiet streams before their first heartbeat. See Bun's timeout guidance. Apply equivalent idle-timeout and buffering settings to any reverse proxy.

Long-lived streams consume host connections and memory. Measure concurrent connections, Redis connection counts, renewal/refetch traffic, latency, and cost in your deployment before increasing capacity. Avoid per-request worker loops and do not use memory as a cross-instance production transport.

Provider instrumentation uses the broadcast watcher for publications, subscriptions, readiness, denials, temporary failures, and continuity loss. Use client status callbacks to observe browser reconnections. Payloads and origin IDs are omitted from broadcast-specific records. Keep application logging equally selective. beignet routes, beignet map, beignet explain, and beignet doctor inspect declarations and wiring; static checks cannot prove policy correctness, transaction timing, or host suitability.