@dxv-systems/realtime

Realtime change signals over AWS AppSync Events: subscribe-token minting, SigV4 publish, and a plain-React subscription hook with a polling fallback.

npm i @dxv-systems/realtime

Principle: signals, not data

Publish a small "what went stale" payload and let the client re-read through its normal, permission-checked path (router.refresh(), invalidateQueries). The realtime layer never needs your data model. A socket only shortens the delay; polling still bounds it.

Requirements

You need an AppSync Events API. Subscribers authenticate with an HS256 JWT minted by mintSubscribeToken. The token is sent as-is (no Bearer prefix) as the Authorization value in the WebSocket connect header and in every subscribe message. Your API's authorizer must verify that token with the same signing secret and allow a channel only when matchChannel(claims.ch, channel) is true. Publishing uses IAM (SigV4).

Environment variables

Set all of these to enable realtime. Every server function reads them lazily, at call time, so your app builds without them and unconfigured environments keep working: publish becomes a no-op and mintSubscribeToken returns null.

Server (./server, Node only)

import { mintSubscribeToken, realtimeClientConfig, publish } from "@dxv-systems/realtime/server";

• mintSubscribeToken({ sub, channel, ttlSeconds? }): HS256 JWT built with node:crypto (no jsonwebtoken dependency). Claims: sub, ns (from REALTIME_NAMESPACE), ch (the channel you pass, which may be a pattern such as /myapp/*), iat, exp (at most iat + 900, default 900). Throws if the first segment of channel is not ns: that is a caller bug, not a runtime condition. Returns null when the env vars are unset.
• realtimeClientConfig({ sub, channel }): the full { token, expiresAt, httpEndpoint, wsEndpoint } your token route returns to the browser, or null. expiresAt is epoch milliseconds, not the JWT's second-precision exp, so the client can compare it to Date.now() directly.
• publish(channel, payload): SigV4 POST /event, 2s timeout, never throws. Returns { ok: true, skipped: true } without sending when the publish env vars are unset. Body: { channel, events: [JSON.stringify(payload)] }.
• publishMany(channels, payload): fan-out convenience. Returns the first failure, or { ok: true, skipped: true } only if every channel skipped.

./server also re-exports matchChannel.

Token route (Next.js App Router)

// app/api/realtime/token/route.ts
import { NextResponse } from "next/server";
import { realtimeClientConfig } from "@dxv-systems/realtime/server";
import { getSession } from "@/lib/session";

export async function GET() {
  const session = await getSession();
  if (!session) return new NextResponse(null, { status: 401 });

  const config = realtimeClientConfig({ sub: session.userId, channel: "/myapp/*" });
  return NextResponse.json(config); // null when realtime isn't configured: the client just doesn't connect
}

Publish after commit (any Node server)

import { publish } from "@dxv-systems/realtime/server";

await db.transaction(async (tx) => { /* write */ });
const result = await publish("/myapp/changes", { type: "changed", paths: ["/dashboard/planning"] });
if (!result.ok) console.error("realtime publish failed", result.error); // never fails the request

Isomorphic (.)

import { matchChannel } from "@dxv-systems/realtime";

matchChannel(pattern, channel) is the channel rule your authorizer should enforce. A pattern without /* matches only that exact channel. A pattern ending in /* matches every channel under that prefix. No Node or DOM dependency, so it is safe in any bundle target.

React (./react, "use client", peer dependency react >=18)

import { useRealtimeChannel, useLiveRefresh, defaultIsBusy } from "@dxv-systems/realtime/react";

useRealtimeChannel(channel, onEvent, { getConfig, onResync? })

Implements the AppSync Events WebSocket protocol. It connects to wss://<wsEndpoint>/event/realtime with the aws-appsync-event-ws and base64url header-... subprotocols, sends connection_init, subscribes on connection_ack, and re-arms a keepalive watchdog on every ka (it closes and reconnects if none arrives within connectionTimeoutMs). It calls onEvent with the parsed payload on every data message.

connected reflects a live subscription (subscribe_success), not just an open socket. Reconnects use capped exponential backoff with jitter (1s to 30s). onEvent, getConfig and onResync are held in refs, so you do not need to memoize them.

Events published between a disconnect and the next subscribe_success are lost: the server cannot redeliver them. onResync fires on every subscribe_success after the first one for that hook instance (every reconnect), so you can re-read state and close that gap. Wire it to the same refresh as onEvent:

useRealtimeChannel(channel, onEvent, { getConfig, onResync: () => router.refresh() });

The socket is rotated after one hour (MAX_CONNECTION_AGE_MS) with a planned reconnect and no backoff. This is not a token refresh: the authorizer only runs on connect and subscribe, never on an open subscription, so a token reaching its 15-minute exp does not need a new socket. The one-hour cap bounds how long a subscription can outlive a rotated signing secret. getConfig is called again on every connect, so each new connection carries a current token.

getConfig returning null means "realtime is not configured here". The hook stays paused (connected: false, no socket, no error) and retries getConfig every 30s. A subscribe_error or broadcast_error logs once via console.warn (with the error payload) before closing and backing off, so a wrong channel pattern shows up in the console.

Shared connection: one WebSocket per app, not per hook. Each hook registers a subscriber on its channel. The first subscriber for a channel sends subscribe, the last sends unsubscribe, and the socket closes only when no channel has subscribers left. Two hooks on the same channel share one subscription and both receive its events. The second one is connected immediately and gets no onResync, because it has missed nothing. A subscribe_error still closes the whole socket (the usual cause is an expired token, and only a reconnect fetches a new one), so every channel re-subscribes together and every hook that was already subscribed is resynced. Subscription ids are assigned per connection in registration order (s1, s2, ...). __resetRealtimeConnection() is exported for tests only: it drops the module-level connection so each test starts clean.

Combining both hooks

useLiveRefresh needs connected from useRealtimeChannel, while useRealtimeChannel's callbacks want requestRefresh from useLiveRefresh. Break the cycle with a ref: call useRealtimeChannel first, route its callbacks through a ref, and fill that ref from useLiveRefresh's return value.

"use client";
import { useRef } from "react";
import { useRouter, usePathname } from "next/navigation";
import { useRealtimeChannel, useLiveRefresh } from "@dxv-systems/realtime/react";

export function LiveRefresh() {
  const router = useRouter();
  const pathname = usePathname();
  const requestRefreshRef = useRef<() => void>(() => {});

  function matches(payload: unknown) {
    const paths = (payload as { paths?: string[] })?.paths ?? [];
    return paths.some((p) => pathname.startsWith(p));
  }

  const { connected } = useRealtimeChannel(
    "/myapp/*",
    (e) => {
      if (matches(e)) requestRefreshRef.current();
    },
    {
      getConfig: () => fetch("/api/realtime/token").then((r) => (r.ok ? r.json() : null)),
      onResync: () => requestRefreshRef.current(), // a reconnect may have missed events
    },
  );

  const { requestRefresh } = useLiveRefresh({
    refresh: () => router.refresh(),
    intervalMs: 2 * 60_000,
    connectedIntervalMs: 10 * 60_000,
    connected,
    // isBusy omitted: the built-in heuristic is used
  });
  requestRefreshRef.current = requestRefresh;

  return null;
}

Vite / TanStack Query

import { useRef } from "react";
import { useQueryClient } from "@tanstack/react-query";
import { useRealtimeChannel, useLiveRefresh } from "@dxv-systems/realtime/react";

function useOrderEvents(orgId: string, orderId: string) {
  const queryClient = useQueryClient();
  const invalidate = () => queryClient.invalidateQueries({ queryKey: ["order", orderId] });

  const requestRefreshRef = useRef<() => void>(() => {});

  const { connected } = useRealtimeChannel(
    `/myapp/org-${orgId}/order-${orderId}`,
    () => requestRefreshRef.current(),
    {
      getConfig: () => fetch("/api/realtime/token").then((r) => (r.ok ? r.json() : null)),
      onResync: () => requestRefreshRef.current(),
    },
  );

  const { requestRefresh } = useLiveRefresh({
    refresh: invalidate,
    intervalMs: 30_000,
    connectedIntervalMs: 2 * 60_000,
    connected,
  });
  requestRefreshRef.current = requestRefresh;
}

useLiveRefresh({ refresh, intervalMs, connectedIntervalMs, connected, isBusy? }) returns { requestRefresh }

Calls refresh() every connected ? connectedIntervalMs : intervalMs, only while the tab is visible. When the tab becomes visible again, it refreshes immediately if the last refresh is older than 30s. While the user is busy it skips a tick and schedules a single retry 15s later (not a stack of retries).

requestRefresh() runs the same gated path on demand (visibility check, busy check with its single 15s retry, last-refresh-time update), so a realtime event can trigger a refresh without bypassing the busy gate. Its identity is stable across renders, so you can pass it directly as onEvent or onResync. Calls within 500ms of each other coalesce into one trailing refresh, so several events in a row cause one refresh. A call made while the tab is hidden is deferred and runs when the user returns to the tab, regardless of the 30s staleness rule.

Default busy check (omit isBusy): a held-down pointer always counts as busy (treated as stale after 5 minutes, as a safety net against a stuck flag). Otherwise, a focused input, textarea, select or contenteditable, or an open dialog ([role="dialog"] or [aria-modal="true"], which also matches Radix dialogs), counts as busy only within 60s of the last keydown, input, pointerdown or wheel. An idle wall display with stale focus in a search box keeps refreshing. The pure decision function is exported as shouldSkipRefresh(facts: BusyFacts) for testing. useLiveRefresh registers its interaction listeners on mount and removes them on unmount; nothing attaches at import time.

Pass isBusy to replace this with your own () => boolean. defaultIsBusy() is exported as a stateless check for use outside useLiveRefresh. It has no interaction-window tracking (a plain function cannot own listeners), so prefer omitting isBusy over passing defaultIsBusy to useLiveRefresh.

Module format

Output is CommonJS. The "use client" directive in ./react survives the build but lands after the "use strict" that the CommonJS output adds. Wrap ./react in your own "use client" component; do not rely on the package's directive to set the boundary.

Protocol notes

• data message event field. The AppSync protocol reference shows a data message's event as a one-element array of a JSON string, while the publish docs describe a plain array of stringified events. The hook accepts both a bare string and an array, then JSON.parses each item.
• Publish headers. publish() sends content-type: application/json plus the SigV4 headers that aws4fetch adds (host, x-amz-date, Authorization), matching the Events HTTP publish docs.
• No publish retries. aws4fetch retries 5xx and 429 responses up to 10 times by default, which would exceed the 2s publish budget. publish() disables those retries.
• No token-driven reconnect. The socket reconnects only on a network drop, a keepalive timeout, a subscribe_error, or the one-hour cap. Closing early to refresh the token would only create a window in which events are lost.