@finueva/pub-sub

Use @finueva/pub-sub to receive live events, manage browser push, issue tickets, and publish events.

Acknowledged Web connections can opt into connect({ deliveryMode: "acknowledged", reauthorization: true, onEvent }) after the broker supports in-place renewal. The SDK obtains fresh tickets through the same getTicket provider, asking for action reauthorize (sign it like connect, with the same capabilities and lease key) and reauthorizes the existing subscription. The Application must freshly verify session authority on every ticket request. Successful renewal emits no new open state; an actual failure closes the socket and still requires Application reconnect/catch-up. Deploy broker support before enabling this option. The default bounded replacement behavior and legacy v1 are unchanged.

Pub Sub is experimental. It attempts delivery immediately. It does not retain, replay, order, retry, deduplicate, or acknowledge events.

Install the SDK

pnpm add @finueva/pub-sub

Node.js tools and server applications require Node.js 22 or later.

Opaque channel signaling

The next minor SDK adds client.createSignalingClient(...) to the canonical createPubSubClient result. It uses the same configured service origin and WebSocket factory for Cloudflare or an explicitly composed Node relay. The trusted-server client's issueChannelTicket({ userId, channel, publish, subscribe, expiresAt }) supplies a single-use grant after Application authorization; expiresAt is an absolute millisecond deadline at most 60 seconds ahead.

const signaling = await client.createSignalingClient({
  userId: authorizedOwner,
  channel: authorizedChannel,
  getTicket: (signal) => requestApplicationChannelTicket(signal),
  onPeers: (self, peers) => peerProvider.updatePeers(self, peers),
  onEvent: ({ peerId, payload }) => peerProvider.receive(peerId, payload),
});
signaling.publish(opaquePayload, optionalRecipientPeerId);
await signaling.closed;

userId is the room's owner namespace, not the connecting participant's personal identity. Participants authorized for that same owner/channel receive distinct signed peerId connection identities. Other owners with the same channel name are isolated. The current owner-room capacity is four connections. Data and peer presence end at capability expiry; the Application must reauthorize and reconnect. Pub Sub does not interpret the payload or implement WebRTC or resource policy.

connectChannel is also available from @finueva/pub-sub/channels. For a server-declared alternate Node route, pass its advertised absolute websocketUrl; it must share the configured service origin and contain no query, fragment or credentials. The server verifies the ticket's signed exact path. Both runtimes use channel/ticket subprotocols, never a ticket query parameter. See the repository's channel architecture for the Node provider contract and route composition. Consume the published minor release after service rollout; source availability does not mean these methods exist in an older installed SDK.

API summary

Browser client:

Server client:

Create a browser client

Provide the authenticated user's ID and a getTicket function. The function must request a ticket from your application server.

import { createPubSubClient } from "@finueva/pub-sub";

const pubSub = await createPubSubClient({
  userId: currentUser.id,
  async getTicket({ action, signal }) {
    const response = await fetch("/api/pub-sub/ticket", {
      method: "POST",
      credentials: "include",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ action }),
      signal,
    });

    if (!response.ok) throw new Error("Pub Sub ticket unavailable");
    return ((await response.json()) as { ticket: string }).ticket;
  },
});

Your server must get the user ID from authenticated session data. Do not accept a user ID from the request body. Never send a Pub Sub API token to browser code.

Production is the default environment. Without explicit public configuration, client creation makes one GET to /v1/configuration on the embedded production service origin, without cookies or Vault headers. It accepts a complete public bundle within three seconds and 16 KiB; redirects, failures, and invalid or partial responses preserve the complete embedded bundle without retry. Passing serviceOrigin, variables, or push.firebase skips this request and merges caller values over the selected embedded bundle. Pass environment: "staging" to select the staging service and fallback bundle. The service returns its cached configuration snapshot, not a fresh OpenBao read. Browser access requires an origin allowed by the service.

Receive live events

Call connect() and process events with onEvent.

const connection = await pubSub.connect({
  onEvent(event) {
    if (event.type === "order.updated") {
      refreshOrder(event.data);
    }
  },
});

// Close the connection on sign-out or account change.
connection.close();
await connection.closed;

event.data is unknown. Validate it before use. The SDK does not reconnect. Call connect() again to create a new connection. Events sent while disconnected are not replayed.

Registered hosted applications must set deliveryMode: "acknowledged" in connect(). This selects finueva.pubsub.v2 without falling back to v1. The SDK validates each bounded frame and admits it to the callback before returning matching transport credit. A synchronous callback failure closes the connection without credit. Credit confirms transport admission, not asynchronous persistence or message read status.

The hosted service closes slow connections when outstanding frame, byte, attachment or deadline bounds are reached. Recover current authorized state through the application's API after reconnect or a cursor gap. No live event body is retained for transport replay. Independent legacy application namespaces retain their separate existing protocol.

Share one connection per browser

With connectionSharing, every tab's acknowledged connect() joins one socket per user in the browser, instead of each tab holding one of the broker's eight per user. The connect() API and the returned connection are unchanged.

import PubSubWorker from "@finueva/pub-sub/shared-connection-worker?sharedworker";

const pubSub = await createPubSubClient({
  userId,
  getTicket: async ({ action, signal, leaseKey }) => myTicket({ action, leaseKey }, signal),
  connectionSharing: { sharedWorker: () => new PubSubWorker({ name: "pub-sub" }) },
});

• The socket runs in a SharedWorker when you supply one, otherwise in a leader tab elected with Web Locks over a BroadcastChannel. mode: "leader" skips the SharedWorker.
• Without Web Locks and BroadcastChannel, each tab connects alone, exactly as without sharing.
• Each tab still proves its own session with a ticket from its own getTicket. A ticket for another user is refused, and different users never share a socket.
• Every tab receives every event of its user. An event is acknowledged once every tab's onEvent returned.
• A leader handoff, leader loss or worker stop reaches each tab as an ordinary closed state, so reconnect and catch-up stay yours.
• A shared connection passes leaseKey to getTicket for its connect tickets and exposes it as connection.leaseKey. Sign it into the ticket and use it for socket-bound presence watches.

See ADR 0006 (one shared user connection per browser) for the design and its trade-offs.

Enable browser push

Install the listener in the service worker that controls your application origin.

import { installPubSubServiceWorker } from "@finueva/pub-sub/service-worker";

installPubSubServiceWorker({
  notificationIcon: "/notification.png",
  openPath: "/inbox",
});

Call enablePush() after a user action, such as a button click.

const stopPushEvents = pubSub.onPush((event) => {
  if (event.kind === "tapped") {
    openEvent(event.eventId, event.eventType);
  }
});

const enabled = await pubSub.enablePush();

// Disable push on opt-out, sign-out, or account change.
await pubSub.disablePush();
await enabled.unlisten();
await stopPushEvents();

Push events contain an event ID, an event type, optional display text, and optionally a tag and routing data entries that your server attached. Fetch current application data from an authorized API after the application opens.

The service-worker helper also accepts an onPush(metadata) hook that runs before display and returns "default" or "handled"; a throw or rejection falls back to "default". When openPath is set, a tap navigates an existing same-origin window to it. By default the SDK keeps its installation ID and token in localStorage under keys that are not scoped to a user; pass a user-scoped push.storage if several users share a browser profile.

Create a server client

Import the server entry point only from server code.

import { createPubSubServerClient } from "@finueva/pub-sub/server";

const pubSub = createPubSubServerClient({
  apiToken: process.env.PUB_SUB_API_TOKEN!,
});

Issue a ticket

const ticket = await pubSub.issueTicket({
  userId: authenticatedUser.id,
  action: "connect",
});

Manage a registered push installation

Registered installation writes may omit idempotency fields for compatibility; each such write is a new server-issued installation incarnation and has no replay receipt. Migration-aware callers should persist a UUIDv7 idempotencyKey, a fresh incarnation UUID, and expiresAt before the first request, then reuse that exact tuple on every retry. The fixed expiry must be seven days after the key's embedded timestamp. The result includes immutable eviction pairs (installationId plus incarnation) and a separately recomputed authority status (owned, unowned, or opaque superseded).

Use observeInstallation() to check token ownership before repair or fallback. Server-client deleteInstallation() is conditional: pass the token and expected incarnation; it calls the separate /authority endpoint and returns removed, absent, or superseded. The browser/Tauri guest disablePush() remains ticket-based and bodyless on the legacy DELETE route. It drains listeners and invalidates the local provider token even if remote revoke fails, but rejects so callers retain the remote-cleanup uncertainty. It rejects with PUSH_TOKEN_DELETE_FAILED only after the service answered that the installation is removed and the local provider token could not be deleted, so an application can finish an opt-out or sign-out on that code. Any other rejection leaves the service removal unconfirmed.

Publish an event

const result = await pubSub.publish({
  userId: recipient.id,
  event: {
    id: crypto.randomUUID(),
    type: "order.updated",
    occurredAt: new Date().toISOString(),
    data: { orderId: order.id },
    push: {
      title: "Order updated",
      body: "Open the application for details.",
    },
  },
});

Omit push to send the event only to live connections. Keep private information out of push display text because the operating system can show it on a lock screen.

Handle errors

SDK operations throw PubSubError. Use its stable code property to select retry or user-interface behavior.

import { PubSubError } from "@finueva/pub-sub";

try {
  await pubSub.connect({
    onEvent(event) {
      handleApplicationEvent(event);
    },
  });
} catch (error) {
  if (error instanceof PubSubError && error.code === "OPERATION_TIMED_OUT") {
    showRetryAction();
  }
}

Publishing is not idempotent. A retry after an unknown result can deliver the same event more than once.

Learn more

• Web applications
• Server applications
• Delivery behavior
• Security
• Service API

License

MIT

Private presence watches (unpublished candidate)

The server entry point provides registerPresenceWatch, renewPresenceWatch and cancelPresenceWatch. Configure the registered application alias as well as its apiToken. Generate a new watch UUID v4 and generation UUID for each new ordered selection, then retain the returned handle for renew/cancel. The Application must freshly authorize the audience and every selected subject before each operation; Pub Sub registration credentials do not establish organization membership. Returned timestamps are immutable server metadata, not an authorization proof. These methods are server-only; complete Application presence and Node provider parity remain unfinished.

registerPresenceWatch and renewPresenceWatch accept the leaseKey the audience's socket was connected with (signed by issueTicket({ action: "connect", leaseKey })). The answer then carries leaseBinding: socket when the service keeps the lease alive on that socket's in-place reauthorization and ends it fifteen seconds after the last socket holding the key closes, request when you must keep renewing. A bound watch still needs a fresh registration before its fifteen-minute index retention ends, and the Application owns revocation: cancel the watch when membership ends. See the socket-bound presence watch lease architecture page.

setPresenceVisibility submits an ordered batch of at most fifty current Application-owned policy revisions. It requires the registered application alias, refuses changed or partial responses, and never accepts a caller clock. leaseMs (60,000 to 900,000) asks for a longer lease; use it only when every policy transition is written as a new revision at once, because a repeated revision never shortens a live longer lease. A failed multi-shard operation may leave bounded grants on successful shards; retry from freshly read policy and retain the Application's disclosure checks. Missing or expired controls are hidden. This candidate does not establish complete Application presence or Node parity.

wakePresencePolicy submits the Application's current canonical policy identity: positive sequence, availabilityRevision, dndRevision, a UUID v4 changeId and reason. The Application registration must carry the operator-granted presence_policy_wake capability. A receipt means a live durable obligation was accepted, never delivery. PRESENCE_POLICY_CONFLICT means the command contradicts the retained identity (stale sequence, changed fields without a new sequence, reused change ID or regressed revision); do not retry it, re-read canonical state instead. PRESENCE_POLICY_RETIRED means the identical command's work already completed or expired; recover with a fresh sequence and change ID. PRESENCE_POLICY_RATE_LIMITED means a later retry of the same command may succeed. Unavailable or ambiguous outcomes keep the same command for a bounded retry. The client performs no automatic retries.