FastPubSub SDK JS

TypeScript SDK for FastPubSubNetwork contract v0.1.

The SDK is made for realtime publish/subscribe apps. It publishes binary payloads to tenant + channel and receives messages from channel patterns such as public.#.

Status

This project is a TypeScript port of the Rust SDK.

Working now:

• REST helpers for discovery, ping, and access token creation.
• WebSocket transport for Node.js and browsers.
• Binary WebSocket protocol codec.
• Metadata format.
• Subscription registry.
• Latest message and latest state helpers.
• Filter pipeline and Rust-compatible filter payload formats.
• Compression, encryption, fragmentation, delta, batching, send-rate, bandwidth, and latest-only filters.
• WS application PING/PONG link quality (pingIntervalSec, linkQuality(), event rttMeasured).
• Publish frame v2 with delivery mode (broadcast, deliverOneLowLatency, deliverOneRandom).

Kept for later work:

• WebTransport is a stub, same as the Rust SDK contract v0.1.

Install

npm install @fastpubsub-sdk/client

For local development:

npm install
npm run check
npm run test
npm run build

Basic Flow

import {
  CreateAccessTokenBuilder,
  TenantGrant,
  open
} from "@fastpubsub-sdk/client";

const session = open("globaltest");

const request = new CreateAccessTokenBuilder()
  .createdBy("my-service")
  .expiresAt("+1h")
  .addTenantGrant(
    new TenantGrant(["tenant_1"])
      .allowPub(["public.#"])
      .allowSub(["public.#"])
  )
  .build();

const token = await session.createAccessToken("MT_master_token", request);

const client = await session.webSocket(token.token).build();
const messages = await client.subscribe("tenant_1", "public.#");

await client.publish("tenant_1", "public.chat", "hello");

const message = await messages.recv();
console.log(message?.channel, message?.payload);

Publish Delivery Modes

publish() always sends a legacy v1 frame (broadcast). Use publishWithOptions when you need deliver-one routing:

await client.publishWithOptions("tenant_1", "public.chat", data, {
  delivery: "deliverOneLowLatency"
});

| delivery | Wire | Behaviour | |------------|------|-----------| | "broadcast" | v1 frame (no tag) | All overlay target nodes and all local subscribers | | "deliverOneLowLatency" | v2, mode 1 | One overlay node with lowest inter-node latency among nodes with subscribers; one random local subscribed connection on the edge | | "deliverOneRandom" | v2, mode 2 | One random overlay node with subscribers; one random local subscribed connection |

If no subscribers exist, deliver-one modes do nothing. Overlay routing requires an updated perimeter.

WS Link Quality

HTTP GET /ping during edge selection measures which edge to connect to. After the WebSocket is open, optional application PING/PONG measures RTT on the active connection. The perimeter only answers PONG; it does not track client RTT.

const client = await session.webSocket(token)
  .pingIntervalSec(3)
  .onWebSocketEvent((event) => {
    if (event.type === "rttMeasured") {
      console.log("rtt ms", event.rttMs);
    }
  })
  .build();

const { lastRttMs, medianRttMs } = client.linkQuality(30);

Allowed intervals are 1, 3, and 5 seconds. Omit pingIntervalSec to disable WS ping.

Fast Edge Selection

await session.webSocket(token).build() resolves an edge automatically when needed. It probes up to four bootstrap candidates in parallel, collects competitors for 200 ms after the first measured response, and aborts remaining probes; an unreachable edge cannot hold startup for its full timeout.

Automatic resolution already uses the validated cache. Call resolveEdge only when an app needs to prewarm discovery or display the selected edge:

const session = open("globaltest");
const edge = await session.resolveEdge({
  cache: "prefer",
  cacheTtlMs: 30 * 60_000,
  maxCandidates: 4,
  pingTimeoutMs: 800,
  selectionWindowMs: 200
});

The cache is checked with a short /ping before use; a stale or failed entry automatically falls back to bootstrap discovery. Browser windows persist it in localStorage. For a Worker, Node.js, or custom persistence, pass an EdgeSelectionCache implementation through edgeCache.

Bootstrap and /ping requests add one-shot query parameters to bypass HTTP caches without adding CORS preflight headers. Each timed probe sends one warm-up /ping, then measures the second /ping, so the measured time is a real edge round trip rather than a cached pong response.

ParcelJS

The package is ESM and browser-friendly. In a browser build, the SDK uses globalThis.WebSocket. The Node.js ws package is only used when no browser WebSocket exists.

import { open } from "@fastpubsub-sdk/client";

Parcel can bundle this import for browser apps.

For smaller bundles, import a filter from its own module:

import { CompressedFilter } from "@fastpubsub-sdk/client/filters/compressed";

The package uses "sideEffects": false, and each filter lives in its own ESM module. Bundlers such as Parcel, Rollup, and esbuild can remove filters that are not imported.

Filters

The filter API matches the Rust SDK pipeline. Outbound filters run in reverse registration order. Inbound filters run in registration order.

import { CompressedFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "public.", new CompressedFilter())
  .build();

Filter Rules For Users And AI

Use the same filter list on the sender and receiver for the same tenant + channel prefix.

Register filters from the outside view to the inside view. Outbound filters run in reverse order, inbound filters run in normal order.

For small bundles, import filters from their own modules:

import { CompressedFilter } from "@fastpubsub-sdk/client/filters/compressed";
import { Delta32Filter } from "@fastpubsub-sdk/client/filters/delta";
import { EncryptionFilter } from "@fastpubsub-sdk/client/filters/encryption";

Use wider prefixes only when all child channels need the same filter. For example, public. covers public.chat and public.state.

Compression Example

Use compression for text, JSON, and repeated binary data. Do not use it for already compressed media.

import { CompressedFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "public.chat.", new CompressedFilter())
  .build();

await client.publish("tenant_1", "public.chat.room1", "hello everyone");

Encryption Example

The default algorithm is ChaCha20-Poly1305, like the Rust SDK. AES-GCM uses native WebCrypto when it is selected.

import {
  CryptoKeyManager,
  EncryptionFilter,
  open
} from "@fastpubsub-sdk/client";

const keys = new CryptoKeyManager();

// The key is always 32 bytes.
keys.addCurrent(1, new Uint8Array(32).fill(7));

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "private.", new EncryptionFilter(keys))
  .build();

await client.publish("tenant_1", "private.chat", "secret message");

Use AES-GCM when you want the browser or Node.js native crypto path:

import {
  CryptoKeyManager,
  EncryptionFilter,
  open
} from "@fastpubsub-sdk/client";

const keys = new CryptoKeyManager();
keys.addCurrent(1, new Uint8Array(32).fill(9));

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter(
    "tenant_1",
    "private.",
    EncryptionFilter.withAlgorithm(keys, "aes256-gcm")
  )
  .build();

Fragmentation Example

Use fragmentation when one payload can be larger than a comfortable WebSocket frame. The subscription must include the main channel and the control channel.

import { FragmentFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "files.", new FragmentFilter(64 * 1024, 16 * 1024))
  .build();

// The subscription must cover files.photo and files.photo._contrl.{id}.
const queue = await client.subscribe("tenant_1", "files.photo.#");

await client.publish("tenant_1", "files.photo", new Uint8Array(200_000));
const message = await queue.recv();
console.log(message?.payload.length);

Delta Example

Use delta filters for binary state with a stable layout. Delta32Filter is a good fit for u32, i32, and f32 style data.

import { Delta32Filter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "game.state.", new Delta32Filter(1000))
  .build();

const state = new Uint8Array(4 * 128);

await client.publish("tenant_1", "game.state.player1", state);

// Change only one 32-bit field.
state[4] = 10;
await client.publish("tenant_1", "game.state.player1", state);

Batch Example

Use channel batching for many small messages on the same channel.

import { ChannelBatchFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .filterTimerMode("sdkDefault100Hz")
  .addFilter("tenant_1", "metrics.", ChannelBatchFilter.withMaxBytes(10, 1024))
  .build();

await client.publish("tenant_1", "metrics.mouse", "x=10");
await client.publish("tenant_1", "metrics.mouse", "x=11");
await client.publish("tenant_1", "metrics.mouse", "x=12");

Send Rate Example

Use SendRateFilter when only the latest value matters.

import { SendRateFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .filterTimerMode("lowLatency200Hz")
  .addFilter("tenant_1", "player.position.", SendRateFilter.fromMillis(33))
  .build();

await client.publish("tenant_1", "player.position.42", "x=1");
await client.publish("tenant_1", "player.position.42", "x=2");
await client.publish("tenant_1", "player.position.42", "x=3");

Latest Only Example

Use LatestOnlyFilter when only the newest message per client should reach subscribers. Outbound payloads get a clientId and counter prefix. Inbound payloads with an older or equal counter are dropped.

Use it for high-frequency streams such as player position or sensor data where late packets are useless.

import { LatestOnlyFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "player.position.", LatestOnlyFilter.fromMillis(60_000))
  .build();

await client.publish("tenant_1", "player.position.42", "x=1");
await client.publish("tenant_1", "player.position.42", "x=2");

Inbound messages without a LatestOnlyFilter header follow InvalidPrefixPolicy:

| Policy | Behavior | |--------|----------| | PassThrough (default) | Pass the payload through unchanged | | Drop | Drop the message without an error | | ErrorEventAndDrop | Return a filter error and drop the message |

Change the policy with withInvalidPrefixPolicy:

import { InvalidPrefixPolicy, LatestOnlyFilter } from "@fastpubsub-sdk/client";

LatestOnlyFilter.fromMillis(60_000)
  .withInvalidPrefixPolicy(InvalidPrefixPolicy.Drop);

When a stale message is dropped, the filter emits a warning FilterNotice. Enable onWebSocketEvent on the builder to receive it as filterNotice:

import { open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .onWebSocketEvent((event) => {
    if (event.type === "filterNotice") {
      console.log(`filter ${event.level}: ${event.message}`);
    }
  })
  .build();

For smaller bundles, import the filter from its own module:

import { LatestOnlyFilter } from "@fastpubsub-sdk/client/filters/latest-only";

Bandwidth Limit Example

Use BandwidthLimiterFilter when one prefix must share one byte-per-second limit.

import { BandwidthLimiterFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .filterTimerMode("sdkDefault100Hz")
  .addFilter(
    "tenant_1",
    "video.",
    BandwidthLimiterFilter.withMaxBurst(
      500_000,
      100_000,
      500,
      256,
      2 * 1024 * 1024
    )
  )
  .build();

Debug Log Example

Use DebugLogFilter only for development or diagnostics.

import { DebugLogFilter, open } from "@fastpubsub-sdk/client";

const session = open("globaltest");
const client = await session.webSocket("AT_token")
  .addFilter("tenant_1", "public.", DebugLogFilter.withOptions("debug.log", {
    inbound: true,
    outbound: true,
    writeHex: true
  }))
  .build();

Available filters:

• DummyFilter
• DebugLogFilter
• CompressedFilter
• ChannelBatchFilter
• FragmentFilter
• Delta16Filter, Delta32Filter, Delta64Filter
• SendRateFilter
• LatestOnlyFilter
• BandwidthLimiterFilter
• EncryptionFilter with CryptoKeyManager

Message Reading

subscribe returns a small async queue.

const queue = await client.subscribe("tenant_1", "public.#");

for await (const message of queue) {
  console.log(message.tenant, message.channel, message.matchedPattern);
}

Build Output

The build command writes JavaScript and type declarations into dist/.

Limits

• publishWithOptions encodes delivery in publish frame v2 (0x02 tag). broadcast and publish() keep the legacy v1 frame.
• WS application PING/PONG measures RTT in the SDK (pingIntervalSec, linkQuality(), event rttMeasured). HTTP /ping is only for edge selection.
• deliverOneLowLatency uses overlay latency between nodes; on the local edge it picks one random subscribed connection.
• WebTransport has no real network task yet.
• Delivery success means local SDK/transport acceptance, not confirmed delivery to every remote subscriber.

Changelog

Release notes are in CHANGELOG.md.