createSocketHub

createSocketHub es el núcleo de Socket de Meeko, independiente del transporte. Recibe las definiciones de los canales (defineSocketChannel), el gestor de sesiones y los servicios, y valida todo lo que entra y sale: Origin, la sesión y la autorización del canal en el handshake, y el evento y su payload contra los schemas en cada mensaje. Aplica los límites de tamaño, de búfer y de cola por peer, y reparte los broadcasts a través de un broker (LocalSocketPubSub en una instancia, RedisSocketPubSub o Durable Objects en varias).

El hub no abre sockets: el host hace el upgrade HTTP con su runtime y le pasa cada peer conectado. createSocketUpgradeHandler hace ese puente de forma portable; createServerApp crea el hub solo cuando la aplicación tiene canales (app.socket.hub).

Uso#

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: "Se esperaba { 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: "hola" } }));
console.log(sent); // [ '{"event":"message","payload":{"text":"HOLA"}}' ]

Tipo#

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
}

Opciones#

Valor devuelto#

Un SocketHub:

SocketConnection tiene receive(message) para cada mensaje del cliente, emit(event, payload) y close(code?, reason?). Un mensaje que no es JSON o no tiene event y payload cierra el peer con 1008 y receive rechaza con INVALID_SOCKET_MESSAGE.

Ejemplos#

Broadcast desde una tarea o una 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 };
  },
});

En una aplicación, el autoimport socket envuelve el hub de la aplicación. En un host propio, llama a hub.broadcast("chat", "general", "message", { text: "Mantenimiento en 5 minutos" }).

Varias instancias con 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:" }) });

Un broadcast en una instancia llega a los peers de la misma sala en todas.

Errores comunes#

Un payload que no cumple el schema del servidor#

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

Rechaza con ApiError INVALID_SOCKET_PAYLOAD (400) antes de publicar: lo que sale también se valida, para que el cliente reciba siempre lo que su contrato dice.

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

Un handshake desde otro origen#

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

ApiError INVALID_ORIGIN (403). El navegador envía Origin en los WebSocket, y sin esta comprobación cualquier página podría conectarse con las cookies del usuario.

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

Límites que no son enteros positivos#

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

RangeError: Invalid socket limits, al crear el hub.

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

Notas#

• Portable. No depende del runtime: los adaptadores de Node, Bun, Deno y Cloudflare usan el mismo hub con su propio WebSocket.
• Entrega. Como mucho una vez, sin cola ni reenvío: un peer desconectado pierde los mensajes. Un peer que no lee (búfer lleno o cola del broker llena) se cierra para no frenar al resto de la sala.
• Sesiones. La sesión se carga en el handshake; si se revoca, el peer se cierra al momento. La autorización del canal se ejecuta también en cada entrega.
• Seguridad. Valida los eventos en las dos direcciones; un mensaje inválido cierra el peer con 1008 sin ejecutar ningún handler.

Relacionado#

• createSocketUpgradeHandlerHandshake portable de Socket, que valida la ruta del canal, ejecuta las comprobaciones del hub y conecta al hub el WebSocket que abre el runtime.
• defineSocketChannelDeclara un canal WebSocket de server/socket con los schemas de cada dirección, sus guardas por sala y evento y los handlers de los mensajes del cliente.
• LocalSocketPubSubBroker de broadcasts de Socket dentro de un solo proceso, el que usa el hub por defecto en Node, Bun y Deno; no alcanza a otras instancias.
• RedisSocketPubSubReparte los broadcasts de Socket entre varias instancias o procesos con el pub/sub de Redis, también fragmentado en Redis Cluster.
• createServerAppCrea la aplicación de servidor portable (páginas, API, RPC, Socket, assets, sesiones y middleware) a partir del manifiesto del build y un cargador de módulos.
• WebSocketMeeko como framework full-stack
• Broadcast y varias instanciasMeeko como framework full-stack