crossws - npm Package Compare versions

Comparing version 0.4.6 to 0.4.7

dist/sync.d.mts

+import { PostgresClientLike, RedisClientLike, SyncAdapter, SyncDriver, SyncDriverName, SyncDriverOptions, SyncMessage, broadcastChannel, cluster, decodeEnvelope, encodeEnvelope, pgsql, redis, setupPrimaryCluster, syncDrivers } from "./_chunks/adapter.mjs";
+export { type PostgresClientLike, type RedisClientLike, type SyncAdapter, type SyncDriver, SyncDriverName, SyncDriverOptions, type SyncMessage, broadcastChannel, cluster, decodeEnvelope, encodeEnvelope, pgsql, redis, setupPrimaryCluster, syncDrivers };

dist/sync.mjs

+function broadcastChannel(opts) {
+	return ({ id }) => {
+		const channel = new BroadcastChannel(opts.channel);
+		return {
+			subscribe(deliver) {
+				channel.addEventListener("message", (event) => {
+					const envelope = event.data;
+					const msg = envelope?.msg;
+					if (!envelope || envelope.id === id || typeof msg?.topic !== "string" || typeof msg.namespace !== "string" || !(typeof msg.data === "string" || msg.data instanceof Uint8Array)) return;
+					deliver(msg);
+				});
+			},
+			publish(msg) {
+				channel.postMessage({
+					id,
+					msg
+				});
+			},
+			close() {
+				channel.close();
+			}
+		};
+	};
+}
+function encodeEnvelope(id, msg) {
+	const binary = msg.data instanceof Uint8Array;
+	return JSON.stringify({
+		id,
+		msg: {
+			namespace: msg.namespace,
+			topic: msg.topic,
+			binary,
+			data: binary ? toBase64(msg.data) : msg.data
+		}
+	});
+}
+function decodeEnvelope(raw) {
+	try {
+		const parsed = JSON.parse(raw);
+		if (!parsed || typeof parsed.id !== "string" || !parsed.msg || typeof parsed.msg.topic !== "string" || typeof parsed.msg.namespace !== "string" || typeof parsed.msg.data !== "string") return;
+		return {
+			id: parsed.id,
+			msg: {
+				namespace: parsed.msg.namespace,
+				topic: parsed.msg.topic,
+				data: parsed.msg.binary ? fromBase64(parsed.msg.data) : parsed.msg.data
+			}
+		};
+	} catch {
+		return;
+	}
+}
+function toBase64(data) {
+	let binary = "";
+	for (const byte of data) binary += String.fromCharCode(byte);
+	return btoa(binary);
+}
+function fromBase64(data) {
+	const binary = atob(data);
+	const out = new Uint8Array(binary.length);
+	for (let i = 0; i < binary.length; i++) out[i] = binary.charCodeAt(i);
+	return out;
+}
+function redis(opts) {
+	const channel = opts.channel;
+	const isNodeRedis = opts.connector === void 0 ? typeof opts.client.pSubscribe === "function" : opts.connector === "node-redis";
+	return ({ id }) => {
+		const subscriber = opts.client.duplicate();
+		let started = false;
+		let onMessage;
+		return {
+			async subscribe(deliver) {
+				started = true;
+				const handle = (raw) => {
+					const envelope = decodeEnvelope(raw);
+					if (!envelope || envelope.id === id) return;
+					deliver(envelope.msg);
+				};
+				if (isNodeRedis) {
+					await subscriber.connect?.();
+					await subscriber.subscribe(channel, (raw) => handle(raw));
+				} else {
+					onMessage = (ch, raw) => {
+						if (ch === channel) handle(raw);
+					};
+					subscriber.on?.("message", onMessage);
+					await subscriber.subscribe(channel);
+				}
+			},
+			async publish(msg) {
+				await opts.client.publish(channel, encodeEnvelope(id, msg));
+			},
+			async close() {
+				if (!started) return;
+				if (onMessage) {
+					subscriber.off?.("message", onMessage);
+					onMessage = void 0;
+				}
+				try {
+					await subscriber.quit?.();
+				} catch (error) {
+					console.error("[crossws] sync redis close failed:", error);
+				}
+			}
+		};
+	};
+}
+function pgsql(opts) {
+	const channel = opts.channel;
+	if (new TextEncoder().encode(channel).length > 63) throw new Error(`[crossws] pgsql sync channel name must be at most 63 bytes (got ${channel.length} chars): ${channel}`);
+	const isPostgresJs = opts.connector === void 0 ? typeof opts.client.listen === "function" : opts.connector === "postgres.js";
+	if (!isPostgresJs && "idleCount" in opts.client && "totalCount" in opts.client && typeof opts.client.query === "function") throw new Error("[crossws] pgsql sync requires a dedicated `Client`, not a `Pool` (pool connections rotate, so LISTEN/NOTIFY can't deliver). Pass `new Client()` (node-postgres) or a `postgres()` instance (postgres.js).");
+	const quotedChannel = `"${channel.replace(/"/g, "\"\"")}"`;
+	return ({ id }) => {
+		let unlisten;
+		let onNotification;
+		return {
+			async subscribe(deliver) {
+				const handle = (raw) => {
+					const envelope = decodeEnvelope(raw);
+					if (!envelope || envelope.id === id) return;
+					deliver(envelope.msg);
+				};
+				if (isPostgresJs) unlisten = (await opts.client.listen(channel, (payload) => handle(payload)))?.unlisten;
+				else {
+					onNotification = (msg) => {
+						if (msg.channel === channel && msg.payload !== void 0) handle(msg.payload);
+					};
+					opts.client.on("notification", onNotification);
+					await opts.client.query(`LISTEN ${quotedChannel}`);
+				}
+			},
+			async publish(msg) {
+				const payload = encodeEnvelope(id, msg);
+				if (isPostgresJs) await opts.client.notify(channel, payload);
+				else await opts.client.query("SELECT pg_notify($1, $2)", [channel, payload]);
+			},
+			async close() {
+				if (isPostgresJs) await unlisten?.();
+				else if (onNotification) {
+					opts.client.removeListener?.("notification", onNotification);
+					onNotification = void 0;
+					await opts.client.query?.(`UNLISTEN ${quotedChannel}`);
+				}
+			}
+		};
+	};
+}
+const CLUSTER_MESSAGE = "crossws:sync";
+function isClusterEnvelope(message) {
+	return typeof message === "object" && message !== null && typeof message[CLUSTER_MESSAGE] === "string" && typeof message.env === "string";
+}
+let relayInstalled = false;
+async function setupPrimaryCluster() {
+	const cluster = (await import("node:cluster")).default;
+	if (!cluster.isPrimary || relayInstalled) return;
+	relayInstalled = true;
+	cluster.on("message", (_worker, message) => {
+		if (!isClusterEnvelope(message)) return;
+		for (const id in cluster.workers) cluster.workers[id]?.send(message);
+	});
+}
+function cluster(opts) {
+	const channel = opts.channel;
+	return ({ id }) => {
+		const proc = globalThis.process;
+		let onMessage;
+		return {
+			subscribe(deliver) {
+				if (typeof proc?.send !== "function") throw new Error("[crossws] cluster sync must run in a worker forked by node:cluster (process.send is unavailable). Call setupPrimaryCluster() in the primary process and start your server in the workers.");
+				onMessage = (message) => {
+					if (!isClusterEnvelope(message) || message[CLUSTER_MESSAGE] !== channel) return;
+					const envelope = decodeEnvelope(message.env);
+					if (!envelope || envelope.id === id) return;
+					deliver(envelope.msg);
+				};
+				proc.on("message", onMessage);
+			},
+			publish(msg) {
+				proc?.send?.({
+					[CLUSTER_MESSAGE]: channel,
+					env: encodeEnvelope(id, msg)
+				});
+			},
+			close() {
+				if (onMessage) {
+					proc?.off?.("message", onMessage);
+					onMessage = void 0;
+				}
+			}
+		};
+	};
+}
+const syncDrivers = [
+	"broadcastChannel",
+	"redis",
+	"pgsql",
+	"cluster"
+];
+export { broadcastChannel, cluster, decodeEnvelope, encodeEnvelope, pgsql, redis, setupPrimaryCluster, syncDrivers };

sync.d.ts

+export * from "./dist/sync.mjs";
+export { default } from "./dist/sync.mjs";

dist/_chunks/_request.mjs

@@ -65,3 +65,3 @@ const StubRequest = /* @__PURE__ */ (() => {
 		bytes() {
-			return Promise.resolve(new Uint8Array());
+			return Promise.resolve(/* @__PURE__ */ new Uint8Array());
 		}
@@ -68,0 +68,0 @@ formData() {

dist/_chunks/adapter.d.mts

@@ -5,4 +5,366 @@ import { WebSocket } from "./web.mjs";
 }
+/**
+ * A single pub/sub event relayed between crossws instances.
+ *
+ * This is the minimal unit a sync backplane needs to transport so that a
+ * `peer.publish()` (or `adapter.publish()`) on one instance reaches the
+ * subscribers connected to every other instance.
+ */
+interface SyncMessage {
+  /**
+   * Pub/sub namespace (matches {@link Peer.namespace}).
+   *
+   * An empty string means "all namespaces" and mirrors a server-side
+   * `adapter.publish(topic, data)` call without an explicit `namespace`.
+   */
+  namespace: string;
+  /** Channel / topic name. */
+  topic: string;
+  /**
+   * Message payload.
+   *
+   * crossws normalizes payloads to a string or `Uint8Array` before handing
+   * them to the driver. Encoding for the wire (e.g. base64 for binary over a
+   * text transport) is the driver's responsibility.
+   */
+  data: string | Uint8Array;
+}
+/**
+ * A live connection to a sync backplane, scoped to one crossws instance.
+ *
+ * Created by a {@link SyncAdapter}. crossws calls {@link SyncDriver.publish}
+ * for every local publish and expects {@link SyncDriver.subscribe} to invoke
+ * the supplied `deliver` callback for every message originating from *other*
+ * instances.
+ */
+interface SyncDriver {
+  /**
+   * Start receiving messages relayed from other instances.
+   *
+   * The driver MUST call `deliver` for every remote message; crossws then
+   * fans it out to local subscribers. The driver MUST NOT echo this
+   * instance's own publishes back (backplanes like Redis pub/sub echo by
+   * default — use the instance `id` from {@link SyncAdapter} to filter).
+   */
+  subscribe(deliver: (message: SyncMessage) => void): MaybePromise<void>;
+  /** Relay a locally-published message to the other instances. */
+  publish(message: SyncMessage): MaybePromise<void>;
+  /** Optional teardown when the adapter shuts down. */
+  close?(): MaybePromise<void>;
+}
+/**
+ * Factory for a {@link SyncDriver}.
+ *
+ * crossws calls it once per adapter instance and passes a stable random `id`
+ * the driver can use for echo suppression.
+ */
+type SyncAdapter = (ctx: {
+  id: string;
+}) => SyncDriver;
+/**
+ * Zero-dependency sync driver built on [`BroadcastChannel`](https://developer.mozilla.org/en-US/docs/Web/API/BroadcastChannel).
+ *
+ * Bridges instances that share a `BroadcastChannel` registry. On Node.js, Deno
+ * and Bun that registry is scoped to a **single process** (it spans the main
+ * thread and its worker threads), so this is for in-process worker fan-out and
+ * tests — not separate OS processes (e.g. Node `cluster`/PM2 forks), which each
+ * have an isolated registry and will silently not sync. (Deno Deploy is the one
+ * exception: its `BroadcastChannel` spans isolates.)
+ *
+ * For multiple processes, hosts or regions you want a networked driver such as
+ * {@link redis} or {@link pgsql}.
+ *
+ * A `channel` name is required: it scopes the cluster, and a shared default
+ * would risk silently bridging unrelated servers running on the same host.
+ *
+ * @example
+ * ```js
+ * import { broadcastChannel } from "crossws/sync";
+ * const adapter = nodeAdapter({ hooks, sync: broadcastChannel({ channel: "my-app" }) });
+ * ```
+ */
+declare function broadcastChannel(opts: {
+  channel: string;
+}): SyncAdapter;
+/**
+ * Structural subset of a Redis client used by {@link redis}, covering both
+ * [ioredis](https://github.com/redis/ioredis) and
+ * [node-redis](https://github.com/redis/node-redis). Kept structural so crossws
+ * stays dependency-free.
+ *
+ * The two clients shape pub/sub differently and {@link redis} bridges them:
+ * - **ioredis** — `subscribe(channel)` plus a shared `"message"` event whose
+ *   listener receives `(channel, message)`; `duplicate()` returns a ready client.
+ * - **node-redis** — `subscribe(channel, listener)` with an inline listener that
+ *   receives `(message, channel)`; `duplicate()` returns a client you must
+ *   `connect()` first.
+ */
+interface RedisClientLike {
+  publish(channel: string, message: string): unknown;
+  /**
+   * ioredis: `subscribe(channel)`; node-redis: `subscribe(channel, listener)`
+   * (the listener receives `(message, channel)`).
+   */
+  subscribe(channel: string, listener?: (message: string, channel: string) => void): unknown;
+  /** ioredis: shared `"message"` event (listener receives `(channel, message)`). */
+  on?(event: "message", listener: (channel: string, message: string) => void): unknown;
+  /** ioredis: detach the `"message"` listener on {@link SyncDriver.close}. */
+  off?(event: "message", listener: (channel: string, message: string) => void): unknown;
+  /** node-redis: a `duplicate()`d client starts disconnected and must `connect()`. */
+  connect?(): unknown;
+  /** Create a second connection for `SUBSCRIBE` (which blocks the connection). */
+  duplicate(): RedisClientLike;
+  /** Tear down the dedicated subscriber connection on {@link SyncDriver.close}. */
+  quit?(): unknown;
+  /** Present (camelCase) only on node-redis — used for auto-detection. */
+  pSubscribe?: unknown;
+}
+/**
+ * Networked sync driver over Redis pub/sub — the realistic multi-region
+ * backplane. Bring your own client; a dedicated `SUBSCRIBE` connection is
+ * derived from it via `duplicate()` (`SUBSCRIBE` blocks the connection it runs
+ * on, so it can't share the one used for `PUBLISH`).
+ *
+ * Works out of the box with both [ioredis](https://github.com/redis/ioredis)
+ * and [node-redis](https://github.com/redis/node-redis): the flavor is
+ * auto-detected (node-redis exposes camelCase commands such as `pSubscribe`),
+ * with an explicit `connector` escape hatch if detection ever guesses wrong.
+ *
+ * Binary payloads are base64-encoded so they survive Redis's text transport.
+ *
+ * A `channel` name is required: it scopes the cluster, and a shared default
+ * would risk silently bridging unrelated servers on the same Redis instance.
+ *
+ * Reconnect note: ioredis auto-resubscribes its channels after a dropped
+ * connection; node-redis does not restore subscriptions the same way, so a
+ * node-redis-backed instance may stop receiving relayed messages after a
+ * transient outage. Prefer ioredis where connection resilience matters.
+ *
+ * @example
+ * ```js
+ * // ioredis
+ * import Redis from "ioredis";
+ * import { redis } from "crossws/sync";
+ * const adapter = nodeAdapter({ hooks, sync: redis({ client: new Redis(), channel: "my-app" }) });
+ * ```
+ *
+ * @example
+ * ```js
+ * // node-redis
+ * import { createClient } from "redis";
+ * import { redis } from "crossws/sync";
+ * const client = await createClient().connect();
+ * const adapter = nodeAdapter({ hooks, sync: redis({ client, channel: "my-app" }) });
+ * ```
+ */
+declare function redis(opts: {
+  /** Redis client used to `PUBLISH`; a subscriber is `duplicate()`d from it. */client: RedisClientLike; /** Pub/sub channel to relay over. */
+  channel: string;
+  /**
+   * Client flavor. Defaults to auto-detection (node-redis exposes the camelCase
+   * `pSubscribe` command; ioredis does not). Set explicitly to override.
+   */
+  connector?: "ioredis" | "node-redis";
+}): SyncAdapter;
+/**
+ * Structural subset of a PostgreSQL client used by {@link pgsql}, covering
+ * both [node-postgres](https://github.com/brianc/node-postgres) (`pg`) and
+ * [postgres.js](https://github.com/porsager/postgres). Kept structural so
+ * crossws stays dependency-free.
+ *
+ * The two clients shape `LISTEN`/`NOTIFY` differently and {@link pgsql}
+ * bridges them:
+ * - **node-postgres** — issue raw `LISTEN`/`pg_notify` via `query()` and receive
+ *   a shared `"notification"` event whose listener gets `{ channel, payload }`.
+ * - **postgres.js** — dedicated `listen(channel, onnotify)` (resolves to a
+ *   handle with `unlisten()`) and `notify(channel, payload)` helpers.
+ */
+interface PostgresClientLike {
+  /** node-postgres: run `LISTEN` / `SELECT pg_notify(...)` / `UNLISTEN`. */
+  query?(sql: string, values?: unknown[]): unknown;
+  /** node-postgres: shared `"notification"` event for inbound `NOTIFY`s. */
+  on?(event: "notification", listener: (msg: {
+    channel: string;
+    payload?: string;
+  }) => void): unknown;
+  /** node-postgres: detach the `"notification"` listener on {@link SyncDriver.close}. */
+  removeListener?(event: "notification", listener: (msg: {
+    channel: string;
+    payload?: string;
+  }) => void): unknown;
+  /**
+   * postgres.js: subscribe to a channel; resolves to a handle exposing
+   * `unlisten()`. Present only on postgres.js — used for auto-detection.
+   */
+  listen?(channel: string, onnotify: (payload: string) => void): unknown;
+  /** postgres.js: send a `NOTIFY` to a channel. */
+  notify?(channel: string, payload: string): unknown;
+}
+/**
+ * Networked sync driver over PostgreSQL [`LISTEN`/`NOTIFY`](https://www.postgresql.org/docs/current/sql-notify.html)
+ * — a backplane for clusters that already run Postgres and would rather not add
+ * Redis. Bring your own client; unlike Redis `SUBSCRIBE`, Postgres `LISTEN` does
+ * not block the connection, so no `duplicate()` is needed: for node-postgres the
+ * *same* client both listens and notifies. (postgres.js `listen()` internally
+ * reserves its own dedicated connection — that's the client's concern, not ours.)
+ *
+ * Pass a single, dedicated [`Client`](https://node-postgres.com/apis/client), not
+ * a [`Pool`](https://node-postgres.com/apis/pool): pool `query()` runs `LISTEN`
+ * on an arbitrary backend that is then returned to the pool, so notifications
+ * would never reach a stable listener. A `Pool` is detected and rejected. For the
+ * same reason, don't share one client across two `pgsql()` drivers on the same
+ * channel — `close()` issues a single `UNLISTEN` that would silence the others.
+ *
+ * Works out of the box with both [node-postgres](https://github.com/brianc/node-postgres)
+ * (`pg`) and [postgres.js](https://github.com/porsager/postgres): the flavor is
+ * auto-detected (postgres.js exposes a `listen()` helper; node-postgres does
+ * not), with an explicit `connector` escape hatch if detection ever guesses
+ * wrong.
+ *
+ * Binary payloads are base64-encoded so they survive the text `NOTIFY` payload.
+ * Note Postgres caps a `NOTIFY` payload at 8000 bytes — keep relayed messages
+ * small (this is a transport limit, not a crossws one).
+ *
+ * A `channel` name is required: it scopes the cluster, so unrelated servers
+ * don't silently bridge through the same database. It is used verbatim as the
+ * notification channel name.
+ *
+ * @example
+ * ```js
+ * // node-postgres (pg)
+ * import { Client } from "pg";
+ * import { pgsql } from "crossws/sync";
+ * const client = new Client();
+ * await client.connect();
+ * const adapter = nodeAdapter({ hooks, sync: pgsql({ client, channel: "my-app" }) });
+ * ```
+ *
+ * @example
+ * ```js
+ * // postgres.js
+ * import postgresjs from "postgres";
+ * import { pgsql } from "crossws/sync";
+ * const sql = postgresjs();
+ * const adapter = nodeAdapter({ hooks, sync: pgsql({ client: sql, channel: "my-app" }) });
+ * ```
+ */
+declare function pgsql(opts: {
+  /** Connected Postgres client used to both `LISTEN` and `NOTIFY`. */client: PostgresClientLike; /** Notification channel to relay over. */
+  channel: string;
+  /**
+   * Client flavor. Defaults to auto-detection (postgres.js exposes a `listen()`
+   * helper; node-postgres does not). Set explicitly to override.
+   */
+  connector?: "pg" | "postgres.js";
+}): SyncAdapter;
+/**
+ * Install the cluster relay in the **primary** process.
+ *
+ * Node `cluster` workers can't message each other directly — IPC only flows
+ * between each worker and the primary — so the primary must rebroadcast every
+ * worker's relay message to the others. Call this once in your primary process
+ * (before or after forking) so {@link cluster} drivers running in the workers
+ * can reach one another. It is a no-op when called from a worker, so guarding
+ * with `cluster.isPrimary` is optional.
+ *
+ * `node:cluster` is imported lazily so merely importing `crossws/sync` stays
+ * runtime-agnostic (it never loads in workerd/Deno where the module is unused).
+ *
+ * @example
+ * ```js
+ * import cluster from "node:cluster";
+ * import { availableParallelism } from "node:os";
+ * import { setupPrimaryCluster } from "crossws/sync";
+ *
+ * if (cluster.isPrimary) {
+ *   setupPrimaryCluster();
+ *   for (let i = 0; i < availableParallelism(); i++) cluster.fork();
+ * } else {
+ *   // ... start your server with `sync: cluster({ channel: "my-app" })`
+ * }
+ * ```
+ */
+declare function setupPrimaryCluster(): Promise<void>;
+/**
+ * Zero-dependency sync driver over Node.js [`cluster`](https://nodejs.org/api/cluster.html)
+ * worker IPC — bridges forked processes on a **single host** (e.g. Node
+ * `cluster` or PM2 `instances`) without a network backplane.
+ *
+ * This fills the gap left by {@link broadcastChannel}, whose registry is scoped
+ * to one process and silently won't sync across forks. For multiple hosts or
+ * regions you still want a networked driver such as {@link redis} or
+ * {@link pgsql}.
+ *
+ * Requires the relay to be installed in the primary via
+ * {@link setupPrimaryCluster}; the driver itself runs in the workers. Workers
+ * can't message each other directly, so all relay flows worker → primary →
+ * workers. Binary payloads are base64-encoded so they survive default JSON IPC
+ * serialization (no `serialization: "advanced"` needed).
+ *
+ * A `channel` name is required: it scopes the cluster and lets multiple apps
+ * share one process tree without bridging into each other.
+ *
+ * @example
+ * ```js
+ * import cluster from "node:cluster";
+ * import { setupPrimaryCluster, cluster as clusterSync } from "crossws/sync";
+ *
+ * if (cluster.isPrimary) {
+ *   setupPrimaryCluster();
+ *   cluster.fork();
+ *   cluster.fork();
+ * } else {
+ *   const ws = nodeAdapter({ hooks, sync: clusterSync({ channel: "my-app" }) });
+ *   // ... start the server
+ * }
+ * ```
+ */
+declare function cluster(opts: {
+  channel: string;
+}): SyncAdapter;
+declare function encodeEnvelope(id: string, msg: SyncMessage): string;
+declare function decodeEnvelope(raw: string): {
+  id: string;
+  msg: SyncMessage;
+} | undefined;
+/**
+ * Names of the built-in sync drivers exported from `crossws/sync`.
+ *
+ * Useful for automatic integration (e.g. Nitro) that needs to enumerate the
+ * available drivers without importing each one.
+ */
+declare const syncDrivers: readonly ["broadcastChannel", "redis", "pgsql", "cluster"];
+/** Name of a built-in sync driver (see {@link syncDrivers}). */
+type SyncDriverName = (typeof syncDrivers)[number];
+/**
+ * Map from a {@link SyncDriverName} to the options object its driver factory
+ * accepts — derived from the factory signatures so it stays in sync.
+ */
+type SyncDriverOptions = {
+  broadcastChannel: Parameters<typeof broadcastChannel>[0];
+  redis: Parameters<typeof redis>[0];
+  pgsql: Parameters<typeof pgsql>[0];
+  cluster: Parameters<typeof cluster>[0];
+};
 declare const kNodeInspect: unique symbol;
 interface PeerContext extends Record<string, unknown> {}
+interface WaitForDrainOptions {
+  /**
+   * Resolve once {@link Peer.bufferedAmount} drops to or below this many bytes.
+   *
+   * @default 0
+   */
+  threshold?: number;
+  /**
+   * Polling interval (in milliseconds) used to re-check {@link Peer.bufferedAmount}.
+   *
+   * @default 100
+   */
+  pollInterval?: number;
+  /**
+   * Abort the wait (e.g. `AbortSignal.timeout(ms)`). The returned promise
+   * rejects with the signal's `reason`.
+   */
+  signal?: AbortSignal;
+}
 interface AdapterInternal {
@@ -14,2 +376,4 @@ ws: unknown;
   context?: PeerContext;
+  /** Optional sync backplane used to relay publishes to other instances. */
+  sync?: SyncDriver;
 }
@@ -45,2 +409,30 @@ declare abstract class Peer<Internal extends AdapterInternal = AdapterInternal> {
   get topics(): Set<string>;
+  /**
+   * Number of bytes queued for transmission but not yet flushed to the client.
+   *
+   * Use this to apply backpressure: pause sending while it grows past a high
+   * watermark and resume once it drops (or on the `drain` hook). Returns `0` on
+   * adapters that do not expose a buffer signal. Refer to the
+   * [compatibility table](https://crossws.h3.dev/guide/peer#compatibility).
+   */
+  get bufferedAmount(): number;
+  /**
+   * Wait until the send buffer drains to `threshold` bytes (default `0`).
+   *
+   * Resolves immediately when there is no backpressure (or on adapters that do
+   * not expose {@link Peer.bufferedAmount}). Otherwise it polls every
+   * `pollInterval` milliseconds until the buffer drains, also resolving early if
+   * the connection is no longer open so a send loop never hangs on a dropped
+   * client.
+   *
+   * ```ts
+   * for (const chunk of stream) {
+   *   peer.send(chunk);
+   *   if (peer.bufferedAmount > 1024 * 1024) {
+   *     await peer.waitForDrain({ threshold: 256 * 1024 });
+   *   }
+   * }
+   * ```
+   */
+  waitForDrain(opts?: WaitForDrainOptions): Promise<void>;
   abstract close(code?: number, reason?: string): void;
@@ -57,6 +449,21 @@ /** Abruptly close the connection */
   }): number | void | undefined;
-  /** Send message to subscribes of topic */
-  abstract publish(topic: string, data: unknown, options?: {
+  /**
+   * Send a message to subscribers of a topic.
+   *
+   * When a sync backplane is configured, the message is also relayed to the
+   * other crossws instances so their subscribers receive it too.
+   */
+  publish(topic: string, data: unknown, options?: {
     compress?: boolean;
   }): void;
+  /**
+   * Adapter-specific, relay-free local fan-out to subscribers of a topic
+   * (excludes this peer). Implemented by each adapter; used both by
+   * {@link Peer.publish} and by the internal cross-instance delivery path.
+   *
+   * @internal Adapter extension point, not part of the stable public API.
+   */
+  abstract _publish(topic: string, data: unknown, options?: {
+    compress?: boolean;
+  }): void;
   toString(): string;
@@ -153,2 +560,10 @@ [Symbol.toPrimitive](): string;
   }) => MaybePromise<void>;
+  /**
+   * The send buffer has drained after backpressure, so it is safe to resume
+   * sending. Pair with {@link Peer.bufferedAmount} to throttle senders.
+   *
+   * **Note:** Only emitted by adapters that expose a drain signal. Refer to the
+   * [compatibility table](https://crossws.h3.dev/guide/peer#compatibility).
+   */
+  drain: (peer: Peer) => MaybePromise<void>;
   /** An error occurs */
@@ -163,3 +578,27 @@ error: (peer: Peer, error: WSError) => MaybePromise<void>;
   }) => void;
+  /**
+   * Gracefully shut the adapter down: close every connected peer (with the
+   * optional `code` / `reason`) and tear down the {@link AdapterInstance.sync}
+   * backplane. Any underlying server you created (e.g. an `http.Server` or a
+   * `WebSocketServer` passed via options) stays yours to close.
+   */
+  readonly close: (code?: number, reason?: string) => Promise<void>;
+  /**
+   * Sync backplane driver, present when an adapter is created with `sync`.
+   *
+   * Closed automatically by {@link AdapterInstance.close}; it leaves any
+   * user-owned client (Redis/Postgres) connected.
+   */
+  readonly sync?: SyncDriver;
 }
+/** Context passed to {@link AdapterOptions.onError} describing what failed. */
+interface SyncErrorContext {
+  /**
+   * Which backplane operation failed:
+   * - `subscribe` — the initial subscription to the backplane.
+   * - `publish` — relaying a local publish out to the other instances.
+   * - `delivery` — fanning an inbound remote message out to local subscribers.
+   */
+  stage: "subscribe" | "publish" | "delivery";
+}
 interface AdapterOptions {
@@ -169,5 +608,20 @@ resolve?: ResolveHooks;
   hooks?: Partial<Hooks>;
+  /**
+   * Optional sync backplane to relay pub/sub between multiple crossws
+   * instances (e.g. across regions/processes). Opt-in: when absent, pub/sub
+   * stays local to the instance, exactly as before.
+   */
+  sync?: SyncAdapter;
+  /**
+   * Called when a {@link AdapterOptions.sync} backplane operation fails.
+   *
+   * Relay is fire-and-forget by design — a flaky backplane never throws into
+   * your `publish` call or crashes the process — so this callback is the only
+   * way to observe a degraded backplane (for logging, metrics or alerting).
+   * Defaults to `console.error`. Has no effect without `sync`.
+   */
+  onError?: (error: unknown, context: SyncErrorContext) => void;
 }
 type Adapter<AdapterT extends AdapterInstance = AdapterInstance, Options extends AdapterOptions = AdapterOptions> = (options?: Options) => AdapterT;
 declare function defineWebSocketAdapter<AdapterT extends AdapterInstance = AdapterInstance, Options extends AdapterOptions = AdapterOptions>(factory: Adapter<AdapterT, Options>): Adapter<AdapterT, Options>;
-export { Adapter, AdapterInstance, AdapterInternal, AdapterOptions, Hooks, Message, Peer, PeerContext, ResolveHooks, WSError, defineHooks, defineWebSocketAdapter };
+export { Adapter, AdapterInstance, AdapterInternal, AdapterOptions, Hooks, Message, Peer, PeerContext, PostgresClientLike, RedisClientLike, ResolveHooks, SyncAdapter, SyncDriver, SyncDriverName, SyncDriverOptions, SyncErrorContext, SyncMessage, WSError, WaitForDrainOptions, broadcastChannel, cluster, decodeEnvelope, defineHooks, defineWebSocketAdapter, encodeEnvelope, pgsql, redis, setupPrimaryCluster, syncDrivers };

dist/_chunks/adapter.mjs

@@ -62,17 +62,94 @@ var AdapterHookable = class {
 }
-function adapterUtils(globalPeers) {
+const kNodeInspect = /*#__PURE__*/ Symbol.for("nodejs.util.inspect.custom");
+function toBufferLike(val) {
+	if (val === void 0 || val === null) return "";
+	const type = typeof val;
+	if (type === "string") return val;
+	if (type === "number" || type === "boolean" || type === "bigint") return val.toString();
+	if (type === "function" || type === "symbol") return "{}";
+	if (val instanceof Uint8Array || val instanceof ArrayBuffer) return val;
+	if (isPlainObject(val)) return JSON.stringify(val);
+	return val;
+}
+function serializeMessage(val) {
+	const data = toBufferLike(val);
+	if (typeof data === "string") return data;
+	return data instanceof Uint8Array ? data : new Uint8Array(data);
+}
+function toString(val) {
+	if (typeof val === "string") return val;
+	const data = toBufferLike(val);
+	if (typeof data === "string") return data;
+	let binary = "";
+	for (const byte of new Uint8Array(data)) binary += String.fromCharCode(byte);
+	return `data:application/octet-stream;base64,${btoa(binary)}`;
+}
+function isPlainObject(value) {
+	if (value === null || typeof value !== "object") return false;
+	const prototype = Object.getPrototypeOf(value);
+	if (prototype !== null && prototype !== Object.prototype && Object.getPrototypeOf(prototype) !== null) return false;
+	if (Symbol.iterator in value) return false;
+	if (Symbol.toStringTag in value) return Object.prototype.toString.call(value) === "[object Module]";
+	return true;
+}
+function adapterUtils(globalPeers, options, caps) {
+	const localPublish = (topic, message, pubOptions) => {
+		for (const peers of pubOptions?.namespace ? [globalPeers.get(pubOptions.namespace) || []] : globalPeers.values()) {
+			let firstPeerWithTopic;
+			for (const peer of peers) if (peer.topics.has(topic)) {
+				firstPeerWithTopic = peer;
+				break;
+			}
+			if (firstPeerWithTopic) {
+				firstPeerWithTopic.send(message, pubOptions);
+				firstPeerWithTopic._publish(topic, message, pubOptions);
+				if (caps?.nativePubSub && !pubOptions?.namespace) break;
+			}
+		}
+	};
+	let sync;
+	if (options?.sync) {
+		const report = (stage, error) => {
+			if (options.onError) options.onError(error, { stage });
+			else console.error(`[crossws] sync ${stage} failed:`, error);
+		};
+		const driver = options.sync({ id: crypto.randomUUID() });
+		const deliver = (msg) => {
+			try {
+				localPublish(msg.topic, msg.data, { namespace: msg.namespace || void 0 });
+			} catch (error) {
+				report("delivery", error);
+			}
+		};
+		try {
+			Promise.resolve(driver.subscribe(deliver)).catch((error) => report("subscribe", error));
+		} catch (error) {
+			report("subscribe", error);
+		}
+		sync = {
+			subscribe: (deliver) => driver.subscribe(deliver),
+			publish: (msg) => {
+				try {
+					return Promise.resolve(driver.publish(msg)).catch((e) => report("publish", e));
+				} catch (error) {
+					report("publish", error);
+				}
+			},
+			close: driver.close ? () => driver.close() : void 0
+		};
+	}
 	return {
 		peers: globalPeers,
+		sync,
 		publish(topic, message, options) {
-			for (const peers of options?.namespace ? [globalPeers.get(options.namespace) || []] : globalPeers.values()) {
-				let firstPeerWithTopic;
-				for (const peer of peers) if (peer.topics.has(topic)) {
-					firstPeerWithTopic = peer;
-					break;
-				}
-				if (firstPeerWithTopic) {
-					firstPeerWithTopic.send(message, options);
-					firstPeerWithTopic.publish(topic, message, options);
-				}
-			}
+			localPublish(topic, message, options);
+			sync?.publish({
+				namespace: options?.namespace || "",
+				topic,
+				data: serializeMessage(message)
+			});
+		},
+		async close(code, reason) {
+			for (const peers of globalPeers.values()) for (const peer of peers) peer.close(code, reason);
+			await sync?.close?.();
 		}
@@ -93,2 +170,2 @@ };
 }
-export { AdapterHookable, adapterUtils, defineHooks, defineWebSocketAdapter, getPeers };
+export { AdapterHookable, adapterUtils, defineHooks, defineWebSocketAdapter, getPeers, kNodeInspect, serializeMessage, toBufferLike, toString };

dist/_chunks/bun.d.mts

@@ -1,2 +0,2 @@
-import { Adapter, AdapterInstance, AdapterOptions, Peer, PeerContext } from "./adapter.mjs";
+import { Adapter, AdapterInstance, AdapterOptions, Peer, PeerContext, SyncDriver } from "./adapter.mjs";
 import { Server, ServerWebSocket, WebSocketHandler } from "bun";
@@ -21,9 +21,11 @@ interface BunAdapter extends AdapterInstance {
   peers: Set<BunPeer>;
+  sync?: SyncDriver;
 }> {
   get remoteAddress(): string;
   get context(): PeerContext;
+  get bufferedAmount(): number;
   send(data: unknown, options?: {
     compress?: boolean;
   }): number;
-  publish(topic: string, data: unknown, options?: {
+  _publish(topic: string, data: unknown, options?: {
     compress?: boolean;
@@ -30,0 +32,0 @@ }): number;

dist/_chunks/node.mjs

@@ -1,4 +0,4 @@
-import { AdapterHookable, adapterUtils, getPeers } from "./adapter.mjs";
 import { import_websocket_server } from "./libs/ws.mjs";
-import { Message, Peer, toBufferLike } from "./peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "./adapter.mjs";
+import { Message, Peer } from "./peer.mjs";
 import { WSError } from "./error.mjs";
@@ -18,2 +18,3 @@ import { StubRequest } from "./_request.mjs";
 	const globalPeers = /* @__PURE__ */ new Map();
+	const baseUtils = adapterUtils(globalPeers, options);
 	const wss = options.wss || new import_websocket_server.default({
@@ -32,3 +33,4 @@ noServer: true,
 			nodeReq,
-			namespace: nodeReq._namespace
+			namespace: nodeReq._namespace,
+			sync: baseUtils.sync
 		});
@@ -46,4 +48,8 @@ peers.add(peer);
 		});
+		const socket = ws._socket;
+		const onDrain = () => hooks.callHook("drain", peer);
+		socket?.on("drain", onDrain);
 		ws.on("close", (code, reason) => {
 			peers.delete(peer);
+			socket?.off("drain", onDrain);
 			hooks.callHook("close", peer, {
@@ -60,3 +66,3 @@ code,
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		handleUpgrade: async (nodeReq, socket, head, webRequest) => {
@@ -96,7 +102,7 @@ const request = webRequest || new NodeReqProxy(nodeReq);
 		});
-		return 0;
+		return this._internal.ws.bufferedAmount;
 	}
-	publish(topic, data, options) {
+	_publish(topic, data, options) {
 		const dataBuff = toBufferLike(data);
-		const isBinary = typeof data !== "string";
+		const isBinary = typeof dataBuff !== "string";
 		const sendOptions = {
@@ -103,0 +109,0 @@ compress: options?.compress,

dist/_chunks/peer.mjs

@@ -1,26 +0,2 @@
-const kNodeInspect = /*#__PURE__*/ Symbol.for("nodejs.util.inspect.custom");
-function toBufferLike(val) {
-	if (val === void 0 || val === null) return "";
-	const type = typeof val;
-	if (type === "string") return val;
-	if (type === "number" || type === "boolean" || type === "bigint") return val.toString();
-	if (type === "function" || type === "symbol") return "{}";
-	if (val instanceof Uint8Array || val instanceof ArrayBuffer) return val;
-	if (isPlainObject(val)) return JSON.stringify(val);
-	return val;
-}
-function toString(val) {
-	if (typeof val === "string") return val;
-	const data = toBufferLike(val);
-	if (typeof data === "string") return data;
-	return `data:application/octet-stream;base64,${btoa(String.fromCharCode(...new Uint8Array(data)))}`;
-}
-function isPlainObject(value) {
-	if (value === null || typeof value !== "object") return false;
-	const prototype = Object.getPrototypeOf(value);
-	if (prototype !== null && prototype !== Object.prototype && Object.getPrototypeOf(prototype) !== null) return false;
-	if (Symbol.iterator in value) return false;
-	if (Symbol.toStringTag in value) return Object.prototype.toString.call(value) === "[object Module]";
-	return true;
-}
+import { kNodeInspect, serializeMessage } from "./adapter.mjs";
 var Message = class {
@@ -150,2 +126,30 @@ event;
 	}
+	get bufferedAmount() {
+		return this._internal.ws?.bufferedAmount ?? 0;
+	}
+	waitForDrain(opts = {}) {
+		const threshold = opts.threshold ?? 0;
+		if (this.bufferedAmount <= threshold) return Promise.resolve();
+		const signal = opts.signal;
+		if (signal?.aborted) return Promise.reject(signal.reason);
+		return new Promise((resolve, reject) => {
+			const check = () => {
+				if (this.bufferedAmount <= threshold || (this.websocket.readyState ?? 1) > 1) {
+					cleanup();
+					resolve();
+				}
+			};
+			const onAbort = () => {
+				cleanup();
+				reject(signal.reason);
+			};
+			const timer = setInterval(check, opts.pollInterval ?? 100);
+			timer.unref?.();
+			const cleanup = () => {
+				clearInterval(timer);
+				signal?.removeEventListener("abort", onAbort);
+			};
+			signal?.addEventListener("abort", onAbort, { once: true });
+		});
+	}
 	terminate() {
@@ -160,2 +164,10 @@ this.close();
 	}
+	publish(topic, data, options) {
+		this._publish(topic, data, options);
+		this._internal.sync?.publish({
+			namespace: this.namespace,
+			topic,
+			data: serializeMessage(data)
+		});
+	}
 	toString() {
@@ -188,2 +200,2 @@ return this.id;
 }
-export { Message, Peer, toBufferLike, toString };
+export { Message, Peer };

dist/adapters/bun.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toBufferLike } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 const bunAdapter = (options = {}) => {
@@ -7,4 +7,5 @@ if (typeof Bun === "undefined") throw new Error("[crossws] Using Bun adapter in an incompatible environment.");
 	const globalPeers = /* @__PURE__ */ new Map();
+	const baseUtils = adapterUtils(globalPeers, options, { nativePubSub: true });
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		async handleUpgrade(request, server) {
@@ -25,3 +26,3 @@ const { upgradeHeaders, endResponse, context, namespace } = await hooks.upgrade(request);
 			message: (ws, message) => {
-				const peer = getPeer(ws, getPeers(globalPeers, ws.data.namespace));
+				const peer = getPeer(ws, getPeers(globalPeers, ws.data.namespace), baseUtils.sync);
 				hooks.callHook("message", peer, new Message(message, peer));
@@ -31,3 +32,3 @@ },
 				const peers = getPeers(globalPeers, ws.data.namespace);
-				const peer = getPeer(ws, peers);
+				const peer = getPeer(ws, peers, baseUtils.sync);
 				peers.add(peer);
@@ -38,3 +39,3 @@ hooks.callHook("open", peer);
 				const peers = getPeers(globalPeers, ws.data.namespace);
-				const peer = getPeer(ws, peers);
+				const peer = getPeer(ws, peers, baseUtils.sync);
 				peers.delete(peer);
@@ -45,2 +46,6 @@ hooks.callHook("close", peer, {
 				});
+			},
+			drain: (ws) => {
+				const peer = getPeer(ws, getPeers(globalPeers, ws.data.namespace));
+				hooks.callHook("drain", peer);
 			}
@@ -50,3 +55,3 @@ }
 };
-function getPeer(ws, peers) {
+function getPeer(ws, peers, sync) {
 	if (ws.data.peer) return ws.data.peer;
@@ -57,3 +62,4 @@ const peer = new BunPeer({
 		peers,
-		namespace: ws.data.namespace
+		namespace: ws.data.namespace,
+		sync
 	});
@@ -70,6 +76,9 @@ ws.data.peer = peer;
 	}
+	get bufferedAmount() {
+		return this._internal.ws.getBufferedAmount();
+	}
 	send(data, options) {
 		return this._internal.ws.send(toBufferLike(data), options?.compress);
 	}
-	publish(topic, data, options) {
+	_publish(topic, data, options) {
 		return this._internal.ws.publish(topic, toBufferLike(data), options?.compress);
@@ -76,0 +85,0 @@ }

dist/adapters/bunny.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toBufferLike } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 import { WSError } from "../_chunks/error.mjs";
@@ -7,4 +7,5 @@ const bunnyAdapter = (options = {}) => {
 	const globalPeers = /* @__PURE__ */ new Map();
+	const baseUtils = adapterUtils(globalPeers, options);
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		handleUpgrade: async (request) => {
@@ -27,3 +28,4 @@ if (!request.upgradeWebSocket || typeof request.upgradeWebSocket !== "function") throw new Error("[crossws] Bunny adapter requires the request to have an upgradeWebSocket method.");
 				peers,
-				context
+				context,
+				sync: baseUtils.sync
 			});
@@ -59,3 +61,3 @@ peers.add(peer);
 	}
-	publish(topic, data) {
+	_publish(topic, data) {
 		const dataBuff = toBufferLike(data);
@@ -62,0 +64,0 @@ for (const peer of this._internal.peers) if (peer !== this && peer._topics.has(topic)) peer._internal.ws.send(dataBuff);

dist/adapters/cloudflare.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toBufferLike } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 import { WSError } from "../_chunks/error.mjs";
@@ -17,3 +17,4 @@ import { StubRequest } from "../_chunks/_request.mjs";
 	});
-	const { publish: durablePublish, ...utils } = adapterUtils(globalPeers);
+	const baseUtils = adapterUtils(globalPeers, opts);
+	const { publish: durablePublish, ...utils } = baseUtils;
 	return {
@@ -69,3 +70,3 @@ ...utils,
 			const server = pair[1];
-			const peer = CloudflareDurablePeer._restore(obj, server, request, namespace);
+			const peer = CloudflareDurablePeer._restore(obj, server, request, namespace, baseUtils.sync);
 			peers.add(peer);
@@ -81,7 +82,7 @@ obj.ctx.acceptWebSocket(server);
 		handleDurableMessage: async (obj, ws, message) => {
-			const peer = CloudflareDurablePeer._restore(obj, ws);
+			const peer = CloudflareDurablePeer._restore(obj, ws, void 0, void 0, baseUtils.sync);
 			await hooks.callHook("message", peer, new Message(message, peer));
 		},
 		handleDurableClose: async (obj, ws, code, reason, wasClean) => {
-			const peer = CloudflareDurablePeer._restore(obj, ws);
+			const peer = CloudflareDurablePeer._restore(obj, ws, void 0, void 0, baseUtils.sync);
 			getPeers(globalPeers, peer.namespace).delete(peer);
@@ -112,3 +113,3 @@ const details = {
 	get peers() {
-		return new Set(this.#getwebsockets().map((ws) => CloudflareDurablePeer._restore(this._internal.durable, ws)));
+		return new Set(this.#getwebsockets().map((ws) => CloudflareDurablePeer._restore(this._internal.durable, ws, void 0, void 0, this._internal.sync)));
 	}
@@ -128,3 +129,3 @@ #getwebsockets() {
 	}
-	publish(topic, data) {
+	_publish(topic, data) {
 		const websockets = this.#getwebsockets();
@@ -141,3 +142,3 @@ if (websockets.length < 2) return;
 	}
-	static _restore(durable, ws, request, namespace) {
+	static _restore(durable, ws, request, namespace, sync) {
 		let peer = ws._crosswsPeer;
@@ -150,3 +151,4 @@ if (peer) return peer;
 			namespace: namespace || state.n || "",
-			durable
+			durable,
+			sync
 		});
@@ -166,3 +168,3 @@ if (state.i) peer._id = state.i;
 	}
-	publish(_topic, _message) {
+	_publish(_topic, _message) {
 		console.warn("[crossws] [cloudflare] pub/sub support requires Durable Objects.");
@@ -169,0 +171,0 @@ }

dist/adapters/deno.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toBufferLike } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 import { WSError } from "../_chunks/error.mjs";
@@ -8,5 +8,8 @@ const denoAdapter = (options = {}) => {
 	const globalPeers = /* @__PURE__ */ new Map();
+	const baseUtils = adapterUtils(globalPeers, options);
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		handleUpgrade: async (request, info) => {
+			const remoteAddress = info.remoteAddr?.hostname;
+			const requestSnapshot = snapshotRequest(request);
 			const { upgradeHeaders, endResponse, context, namespace } = await hooks.upgrade(request);
@@ -22,7 +25,8 @@ if (endResponse) return endResponse;
 				ws: upgrade.socket,
-				request,
+				request: requestSnapshot,
 				peers,
-				denoInfo: info,
+				remoteAddress,
 				context,
-				namespace
+				namespace,
+				sync: baseUtils.sync
 			});
@@ -48,5 +52,14 @@ peers.add(peer);
 };
+function snapshotRequest(request) {
+	const url = request.url;
+	const headers = new Headers(request.headers);
+	return new Proxy(request, { get(target, prop, receiver) {
+		if (prop === "url") return url;
+		if (prop === "headers") return headers;
+		return Reflect.get(target, prop, receiver);
+	} });
+}
 var DenoPeer = class extends Peer {
 	get remoteAddress() {
-		return this._internal.denoInfo.remoteAddr?.hostname;
+		return this._internal.remoteAddress;
 	}
@@ -56,3 +69,3 @@ send(data) {
 	}
-	publish(topic, data) {
+	_publish(topic, data) {
 		const dataBuff = toBufferLike(data);
@@ -59,0 +72,0 @@ for (const peer of this._internal.peers) if (peer !== this && peer._topics.has(topic)) peer._internal.ws.send(dataBuff);

dist/adapters/sse.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toString } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toString } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 const sseAdapter = (opts = {}) => {
@@ -7,4 +7,5 @@ const hooks = new AdapterHookable(opts);
 	const peersMap = opts.bidir ? /* @__PURE__ */ new Map() : void 0;
+	const baseUtils = adapterUtils(globalPeers, opts);
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		fetch: async (request) => {
@@ -35,3 +36,4 @@ const { upgradeHeaders, endResponse, context, namespace } = await hooks.upgrade(request);
 					context,
-					namespace
+					namespace,
+					sync: baseUtils.sync
 				});
@@ -88,3 +90,3 @@ peers.add(peer);
 	}
-	publish(topic, data) {
+	_publish(topic, data) {
 		const dataBuff = toString(data);
@@ -91,0 +93,0 @@ for (const peer of this._internal.peers) if (peer !== this && peer._topics.has(topic)) peer._sendEvent("message", dataBuff);

dist/adapters/uws.d.mts

@@ -1,2 +0,2 @@
-import { Adapter, AdapterInstance, AdapterOptions, Peer, PeerContext } from "../_chunks/adapter.mjs";
+import { Adapter, AdapterInstance, AdapterOptions, Peer, PeerContext, SyncDriver } from "../_chunks/adapter.mjs";
 import { WebSocket } from "../_chunks/web.mjs";
@@ -32,2 +32,3 @@ import uws from "uWebSockets.js";
   uwsData: UserData;
+  sync?: SyncDriver;
 }> {
@@ -41,3 +42,3 @@ get remoteAddress(): string | undefined;
   unsubscribe(topic: string): void;
-  publish(topic: string, message: string, options?: {
+  _publish(topic: string, message: string, options?: {
     compress?: boolean;
@@ -52,4 +53,4 @@ }): number;
 declare class UwsWebSocketProxy implements Partial<WebSocket> {
+  readyState?: number;
   private _uws;
-  readyState?: number;
   constructor(_uws: uws.WebSocket<UserData>);
@@ -56,0 +57,0 @@ get bufferedAmount(): number;

dist/adapters/uws.mjs

@@ -1,3 +0,3 @@
-import { AdapterHookable, adapterUtils, getPeers } from "../_chunks/adapter.mjs";
-import { Message, Peer, toBufferLike } from "../_chunks/peer.mjs";
+import { AdapterHookable, adapterUtils, getPeers, toBufferLike } from "../_chunks/adapter.mjs";
+import { Message, Peer } from "../_chunks/peer.mjs";
 import { StubRequest } from "../_chunks/_request.mjs";
@@ -7,4 +7,5 @@ const uwsAdapter = (options = {}) => {
 	const globalPeers = /* @__PURE__ */ new Map();
+	const baseUtils = adapterUtils(globalPeers, options, { nativePubSub: true });
 	return {
-		...adapterUtils(globalPeers),
+		...baseUtils,
 		websocket: {
@@ -14,3 +15,3 @@ ...options.uws,
 				const peers = getPeers(globalPeers, ws.getUserData().namespace);
-				const peer = getPeer(ws, peers);
+				const peer = getPeer(ws, peers, baseUtils.sync);
 				peer._internal.ws.readyState = 2;
@@ -25,8 +26,12 @@ peers.delete(peer);
 			message(ws, message, _isBinary) {
-				const peer = getPeer(ws, getPeers(globalPeers, ws.getUserData().namespace));
+				const peer = getPeer(ws, getPeers(globalPeers, ws.getUserData().namespace), baseUtils.sync);
 				hooks.callHook("message", peer, new Message(message, peer));
 			},
+			drain(ws) {
+				const peer = getPeer(ws, getPeers(globalPeers, ws.getUserData().namespace));
+				hooks.callHook("drain", peer);
+			},
 			open(ws) {
 				const peers = getPeers(globalPeers, ws.getUserData().namespace);
-				const peer = getPeer(ws, peers);
+				const peer = getPeer(ws, peers, baseUtils.sync);
 				peers.add(peer);
@@ -76,3 +81,3 @@ hooks.callHook("open", peer);
 };
-function getPeer(uws, peers) {
+function getPeer(uws, peers, sync) {
 	const uwsData = uws.getUserData();
@@ -86,3 +91,4 @@ if (uwsData.peer) return uwsData.peer;
 		namespace: uwsData.namespace,
-		uwsData
+		uwsData,
+		sync
 	});
@@ -114,3 +120,3 @@ uwsData.peer = peer;
 	}
-	publish(topic, message, options) {
+	_publish(topic, message, options) {
 		const data = toBufferLike(message);
@@ -145,4 +151,4 @@ const isBinary = typeof data !== "string";
 var UwsWebSocketProxy = class {
+	readyState = 1;
 	_uws;
-	readyState = 1;
 	constructor(_uws) {
@@ -149,0 +155,0 @@ this._uws = _uws;

dist/index.d.mts

@@ -1,2 +0,2 @@
-import { Adapter, AdapterInstance, AdapterInternal, AdapterOptions, Hooks, Message, Peer, PeerContext, ResolveHooks, WSError, defineHooks, defineWebSocketAdapter } from "./_chunks/adapter.mjs";
+import { Adapter, AdapterInstance, AdapterInternal, AdapterOptions, Hooks, Message, Peer, PeerContext, ResolveHooks, SyncAdapter, SyncDriver, SyncErrorContext, SyncMessage, WSError, WaitForDrainOptions, defineHooks, defineWebSocketAdapter } from "./_chunks/adapter.mjs";
 import { ServerWithWSOptions, WSOptions } from "./_chunks/_types.mjs";
@@ -8,5 +8,13 @@ interface WebSocketProxyOptions {
    * Can be a static string/URL or a function that resolves the target dynamically
-   * based on the incoming {@link Peer}.
+   * based on the incoming {@link Peer}. The resolver may be **async** (return a
+   * promise) — useful when the upstream address isn't known yet at connect time
+   * (e.g. a worker that's still booting or being hot-reloaded). Client frames
+   * sent in the meantime are buffered (bounded by {@link maxBufferSize}) and a
+   * non-zero {@link connectTimeout} also covers the resolution, so a resolver
+   * that never settles closes the peer with `1011` rather than hanging. With
+   * `connectTimeout: 0` (timeout disabled) a never-settling resolver leaves the
+   * peer open until {@link maxBufferSize} is hit (`1009`), so pair an unbounded
+   * timeout with your own resolver deadline.
    */
-  target: string | URL | ((peer: Peer) => string | URL);
+  target: string | URL | ((peer: Peer) => string | URL | Promise<string | URL>);
   /**
@@ -98,2 +106,2 @@ * Subprotocol(s) to offer the upstream during the handshake.
 declare function createWebSocketProxy(target: WebSocketProxyOptions["target"] | WebSocketProxyOptions): Partial<Hooks>;
-export { type Adapter, type AdapterInstance, type AdapterInternal, type AdapterOptions, type Hooks, type Message, type Peer, type PeerContext, type ResolveHooks, type ServerWithWSOptions, type WSError, type WSOptions, type WebSocketProxyOptions, createWebSocketProxy, defineHooks, defineWebSocketAdapter };
+export { type Adapter, type AdapterInstance, type AdapterInternal, type AdapterOptions, type Hooks, type Message, type Peer, type PeerContext, type ResolveHooks, type ServerWithWSOptions, type SyncAdapter, type SyncDriver, type SyncErrorContext, type SyncMessage, type WSError, type WSOptions, type WaitForDrainOptions, type WebSocketProxyOptions, createWebSocketProxy, defineHooks, defineWebSocketAdapter };

dist/index.mjs

@@ -19,15 +19,4 @@ import { defineHooks, defineWebSocketAdapter } from "./_chunks/adapter.mjs";
 		open(peer) {
-			let ws;
-			try {
-				const url = _resolveTarget(options.target, peer);
-				const protocols = _resolveProtocols(peer, options.forwardProtocol);
-				const wsOptions = _resolveWsOptions(options.headers, peer);
-				ws = wsOptions ? new WebSocketCtor(url, protocols, wsOptions) : new WebSocketCtor(url, protocols);
-				ws.binaryType = "arraybuffer";
-			} catch {
-				_safeClose(peer, 1011, "Upstream setup failed");
-				return;
-			}
 			const state = {
-				ws,
+				ws: void 0,
 				buffer: [],
@@ -45,22 +34,16 @@ bufferSize: 0,
 			}, timeoutMs);
-			ws.addEventListener("open", () => {
-				_clearTimeout(state);
-				state.open = true;
-				for (const data of state.buffer) ws.send(data);
-				state.buffer.length = 0;
-				state.bufferSize = 0;
-			});
-			ws.addEventListener("message", (event) => {
-				_safeSend(peer, event.data);
-			});
-			ws.addEventListener("close", (event) => {
-				if (upstreams.get(peer.id) !== state) return;
+			let resolved;
+			try {
+				resolved = _resolveTarget(options.target, peer);
+			} catch {
 				_cleanupState(upstreams, peer.id, state);
-				_safeClose(peer, _remapIncomingCode(event.code), event.reason);
-			});
-			ws.addEventListener("error", () => {
+				_safeClose(peer, 1011, "Upstream setup failed");
+				return;
+			}
+			if (resolved instanceof Promise) resolved.then((url) => _dialUpstream(upstreams, peer, state, url, options, WebSocketCtor), () => {
 				if (upstreams.get(peer.id) !== state) return;
 				_cleanupState(upstreams, peer.id, state);
-				_safeClose(peer, 1011, "Upstream error");
+				_safeClose(peer, 1011, "Upstream setup failed");
 			});
+			else _dialUpstream(upstreams, peer, state, resolved, options, WebSocketCtor);
 		},
@@ -73,3 +56,3 @@ message(peer, message) {
 				try {
-					state.ws.send(raw);
+					state.ws?.send(raw);
 				} catch {}
@@ -94,3 +77,3 @@ return;
 			try {
-				state.ws.close(_normalizeOutgoingCode(details.code), _truncateReason(details.reason));
+				state.ws?.close(_normalizeOutgoingCode(details.code), _truncateReason(details.reason));
 			} catch {}
@@ -104,3 +87,3 @@ },
 			try {
-				state.ws.close(1011, "Peer error");
+				state.ws?.close(1011, "Peer error");
 			} catch {}
@@ -110,2 +93,42 @@ }
 }
+function _dialUpstream(upstreams, peer, state, url, options, WebSocketCtor) {
+	if (upstreams.get(peer.id) !== state) return;
+	let ws;
+	try {
+		const protocols = _resolveProtocols(peer, options.forwardProtocol);
+		const wsOptions = _resolveWsOptions(options.headers, peer);
+		ws = wsOptions ? new WebSocketCtor(url, protocols, wsOptions) : new WebSocketCtor(url, protocols);
+		ws.binaryType = "arraybuffer";
+	} catch {
+		_cleanupState(upstreams, peer.id, state);
+		_safeClose(peer, 1011, "Upstream setup failed");
+		return;
+	}
+	state.ws = ws;
+	ws.addEventListener("open", () => {
+		if (upstreams.get(peer.id) !== state) return;
+		_clearTimeout(state);
+		state.open = true;
+		try {
+			for (const data of state.buffer) ws.send(data);
+		} catch {} finally {
+			state.buffer.length = 0;
+			state.bufferSize = 0;
+		}
+	});
+	ws.addEventListener("message", (event) => {
+		if (upstreams.get(peer.id) !== state) return;
+		_safeSend(peer, event.data);
+	});
+	ws.addEventListener("close", (event) => {
+		if (upstreams.get(peer.id) !== state) return;
+		_cleanupState(upstreams, peer.id, state);
+		_safeClose(peer, _remapIncomingCode(event.code), event.reason);
+	});
+	ws.addEventListener("error", () => {
+		if (upstreams.get(peer.id) !== state) return;
+		_cleanupState(upstreams, peer.id, state);
+		_safeClose(peer, 1011, "Upstream error");
+	});
+}
 function _cleanupState(upstreams, id, state) {
@@ -115,3 +138,3 @@ _clearTimeout(state);
 	try {
-		state.ws.close();
+		state.ws?.close();
 	} catch {}
@@ -127,4 +150,8 @@ }
 	const raw = typeof target === "function" ? target(peer) : target;
+	if (_isThenable(raw)) return Promise.resolve(raw).then((value) => value instanceof URL ? value : new URL(value));
 	return raw instanceof URL ? raw : new URL(raw);
 }
+function _isThenable(value) {
+	return value != null && (typeof value === "object" || typeof value === "function") && typeof value.then === "function";
+}
 function _resolveWsOptions(headers, peer) {
@@ -131,0 +158,0 @@ if (!headers) return;

package.json

 {
   "name": "crossws",
-  "version": "0.4.6",
+  "version": "0.4.7",
   "description": "Cross-platform WebSocket Servers for Node.js, Deno, Bun and Cloudflare Workers",
@@ -22,2 +22,3 @@ "homepage": "https://crossws.h3.dev",
     ".": "./dist/index.mjs",
+    "./sync": "./dist/sync.mjs",
     "./adapters/bun": "./dist/adapters/bun.mjs",
@@ -72,10 +73,10 @@ "./adapters/bunny": "./dist/adapters/bunny.mjs",
   "devDependencies": {
-    "@cloudflare/workers-types": "^4.20260608.1",
+    "@cloudflare/workers-types": "^4.20260629.1",
     "@types/bun": "^1.3.14",
     "@types/deno": "^2.7.0",
-    "@types/node": "^25.9.2",
-    "@types/web": "^0.0.350",
+    "@types/node": "^26.0.1",
+    "@types/web": "^0.0.351",
     "@types/ws": "^8.18.1",
-    "@typescript/native-preview": "7.0.0-dev.20260608.1",
-    "@vitest/coverage-v8": "^4.1.8",
+    "@typescript/native-preview": "7.0.0-dev.20260629.1",
+    "@vitest/coverage-v8": "^4.1.9",
     "automd": "^0.4.3",
@@ -91,12 +92,12 @@ "changelogen": "^0.6.2",
     "listhen": "^1.10.0",
-    "obuild": "^0.4.36",
-    "oxfmt": "^0.53.0",
-    "oxlint": "^1.68.0",
-    "srvx": "^0.11.16",
+    "obuild": "^0.4.37",
+    "oxfmt": "^0.56.0",
+    "oxlint": "^1.71.0",
+    "srvx": "^0.11.17",
     "typescript": "^6.0.3",
     "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.57.0",
     "unbuild": "^3.6.1",
-    "undici": "^8.4.1",
-    "vitest": "^4.1.8",
-    "wrangler": "^4.98.0",
+    "undici": "^8.5.0",
+    "vitest": "^4.1.9",
+    "wrangler": "^4.105.0",
     "ws": "^8.21.0"
@@ -115,17 +116,3 @@ },
   },
-  "packageManager": "pnpm@11.5.2",
-  "pnpm": {
-    "ignoredBuiltDependencies": [
-      "@parcel/watcher",
-      "esbuild",
-      "sharp",
-      "workerd"
-    ],
-    "onlyBuiltDependencies": [
-      "@parcel/watcher",
-      "esbuild",
-      "sharp",
-      "workerd"
-    ]
-  }
+  "packageManager": "pnpm@11.7.0"
 }

dist/_chunks/rolldown-runtime.mjs

-import { createRequire } from "node:module";
-var __create = Object.create;
-var __defProp = Object.defineProperty;
-var __getOwnPropDesc = Object.getOwnPropertyDescriptor;
-var __getOwnPropNames = Object.getOwnPropertyNames;
-var __getProtoOf = Object.getPrototypeOf;
-var __hasOwnProp = Object.prototype.hasOwnProperty;
-var __commonJSMin = (cb, mod) => () => (mod || (cb((mod = { exports: {} }).exports, mod), cb = null), mod.exports);
-var __copyProps = (to, from, except, desc) => {
-	if (from && typeof from === "object" || typeof from === "function") for (var keys = __getOwnPropNames(from), i = 0, n = keys.length, key; i < n; i++) {
-		key = keys[i];
-		if (!__hasOwnProp.call(to, key) && key !== except) __defProp(to, key, {
-			get: ((k) => from[k]).bind(null, key),
-			enumerable: !(desc = __getOwnPropDesc(from, key)) || desc.enumerable
-		});
-	}
-	return to;
-};
-var __toESM = (mod, isNodeMode, target) => (target = mod != null ? __create(__getProtoOf(mod)) : {}, __copyProps(isNodeMode || !mod || !mod.__esModule ? __defProp(target, "default", {
-	value: mod,
-	enumerable: true
-}) : target, mod));
-var __require = /* @__PURE__ */ createRequire(import.meta.url);
-export { __commonJSMin, __require, __toESM };

New alerts

Debug access

Supply chain risk

Uses debug, reflection and dynamic code execution features.

Fixed alerts

Debug access

Supply chain risk

Uses debug, reflection and dynamic code execution features.

Improved metrics

Total package byte prevSize

230647

16.43%

Number of package files

73

2.82%

Lines of code

4503

8.61%

No dependency changes