createSocketHub

createSocketHub is Meeko's transport-independent Socket core. It receives channel definitions (defineSocketChannel), the session manager and services, and validates everything coming in and going out: Origin, session and channel authorization at handshake, and the event and its payload against schemas on every message. It applies per-peer size, buffer and queue limits, and distributes broadcasts through a broker (LocalSocketPubSub on one instance, RedisSocketPubSub or Durable Objects on multiple instances).

The hub does not open sockets: the host performs the HTTP upgrade with its runtime and passes each connected peer to it. createSocketUpgradeHandler provides this bridge portably; createServerApp creates the hub automatically when the application has channels (app.socket.hub).

Usage#

import { createSocketHub, defineSocketChannel, MemorySessionStore, SessionManager } from "@meekojs/meeko/server";
import type { Schema, SocketPeer } from "@meekojs/meeko/server";

const text: Schema<{ text: string }> = {
  "~standard": {
    version: 1,
    validate: (value) => typeof value === "object" && value !== null && typeof (value as { text?: unknown }).text === "string"
      ? { value: value as { text: string } }
      : { issues: [{ message: "Expected { text }" }] },
  },
};
const chat = defineSocketChannel({
  clientToServer: { message: text },
  serverToClient: { message: text },
  handlers: { message: ({ payload, emit }) => emit("message", { text: payload.text.toUpperCase() }) },
});

const hub = createSocketHub({ channels: { chat }, sessions: new SessionManager({ store: new MemorySessionStore() }), services: {} });

const sent: string[] = [];
const peer: SocketPeer = { send: (message) => { sent.push(message); }, close: () => {} };
const request = new Request("http://localhost/_meeko/socket/chat?room=general", { headers: { origin: "http://localhost" } });
const connection = await hub.accept("chat", request, peer, "general");
await connection.receive(JSON.stringify({ event: "message", payload: { text: "hello" } }));
console.log(sent); // [ '{"event":"message","payload":{"text":"HELLO"}}' ]

Type#

function createSocketHub<TServices>(options: SocketHubOptions<TServices>): SocketHub

interface SocketHubOptions<TServices = unknown> {
	channels: Readonly<Record<string, unknown>>
	sessions: SessionManager
	services: TServices | ((request: Request) => TServices | Promise<TServices>)
	allowedOrigins?: readonly string[]
	maxMessageBytes?: number
	maxBufferBytes?: number
	maxPendingDeliveries?: number
	sessionRefreshMs?: number
	pubsub?: SocketPubSub
	platform?: ServerPlatform
	onDeliveryError?(error: unknown, context: SocketDeliveryErrorContext): void
}

interface SocketHub {
  preflight(name: string, request: Request, roomId?: string): Promise<void>;
  accept(
    name: string,
    request: Request,
    peer: SocketPeer,
    roomId?: string,
  ): Promise<SocketConnection>;
  broadcast(name: string, roomId: string, event: string, payload: unknown): Promise<void>;
}

interface SocketPeer {
	send(message: string): void | Promise<void>
	close(code?: number, reason?: string): void
	readonly bufferedAmount?: number
}

interface SocketConnection {
	receive(message: string | Uint8Array): Promise<void>
	emit(event: string, payload: unknown): Promise<void>
	close(code?: number, reason?: string): void
}

Options#

Return value#

A SocketHub:

SocketConnection has receive(message) for each client message, emit(event, payload) and close(code?, reason?). A message that is not JSON or lacks event and payload closes the peer with 1008 and receive rejects with INVALID_SOCKET_MESSAGE.

Examples#

Broadcast from a task or API#

// server/api/maintenance.post.ts
export default defineApiRoute({
  auth: "required",
  body: Notice,
  response: Ok,
  handler: async ({ body }) => {
    await socket.chat.broadcast("general", "message", { text: body.text });
    return { ok: true };
  },
});

In an application, the socket auto-import wraps its hub. In a custom host, call hub.broadcast("chat", "general", "message", { text: "Mantenimiento en 5 minutos" }).

Multiple instances with Redis#

import { createClient } from "redis";
import { createSocketHub, RedisSocketPubSub } from "@meekojs/meeko/server";

const publisher = createClient({ url: process.env.REDIS_URL });
const subscriber = publisher.duplicate();
await Promise.all([publisher.connect(), subscriber.connect()]);

const hub = createSocketHub({ channels, sessions, services: {}, pubsub: new RedisSocketPubSub({ publisher, subscriber, prefix: "tienda:prod:socket:" }) });

A broadcast on one instance reaches peers in the same room on all instances.

Common mistakes#

A payload failing the server schema#

await hub.broadcast("chat", "general", "message", { body: "hello" });

Rejects with ApiError INVALID_SOCKET_PAYLOAD (400) before publishing: outgoing data is validated too, so the client always receives what its contract specifies.

await hub.broadcast("chat", "general", "message", { text: "hello" });

A handshake from another origin#

await hub.preflight("chat", new Request("http://localhost/_meeko/socket/chat", { headers: { origin: "https://otro.example" } }));

ApiError INVALID_ORIGIN (403). The browser sends Origin on WebSockets, and without this check any page could connect with the user's cookies.

createSocketHub({ channels, sessions, services: {}, allowedOrigins: ["https://app.example.com"] });

Limits that are not positive integers#

createSocketHub({ channels, sessions, services: {}, maxMessageBytes: 0 });

RangeError: Invalid socket limits, when creating the hub.

createSocketHub({ channels, sessions, services: {}, maxMessageBytes: 16 * 1024 });

Notes#

• Portable. Does not depend on the runtime: Node, Bun, Deno and Cloudflare adapters use the same hub with their own WebSocket.
• Delivery. At most once, without queueing or resending: a disconnected peer loses messages. A peer that does not read (full buffer or broker queue) closes to avoid slowing down the rest of the room.
• Sessions. The session loads at handshake; revocation closes the peer immediately. Channel authorization also runs on each delivery.
• Security. Validates events in both directions; an invalid message closes the peer with 1008 without running any handler.

Related#

• createSocketUpgradeHandlerPortable Socket handshake, validating the channel path, running hub checks and connecting the runtime's WebSocket to the hub.
• defineSocketChannelDeclares a WebSocket channel in server/socket with schemas for each direction, per-room and per-event guards, and handlers for client messages.
• LocalSocketPubSubSocket broadcast broker within a single process, used by the hub by default in Node, Bun and Deno; does not reach other instances.
• RedisSocketPubSubDistributes Socket broadcasts across multiple instances or processes with Redis pub/sub, including sharded pub/sub in Redis Cluster.
• createServerAppCreates the portable server application (pages, API, RPC, Socket, assets, sessions and middleware) from the build manifest and a module loader.
• WebSocketMeeko as a full-stack framework
• Broadcast and multiple instancesMeeko as a full-stack framework