Skip to main content

Realtime hub and transports

createRealtimeHub() separates connection identity, session validation, channel subscriptions, and delivery limits from the transport. It does not start a WebSocket or Socket.IO server.

Creating the hub​

import { createRealtimeHub } from "@kavtuai/guildgate/realtime";

const hub = createRealtimeHub({
  sessions: gate.sessions,
  rateLimits: gate.config.stores.rateLimits,
  allowedOrigins: gate.config.security.allowedOrigins,
  heartbeatIntervalMs: 25_000,
  idleTimeoutMs: 60_000,
  maximumLifetimeMs: 6 * 60 * 60_000,
  maximumMessageBytes: 16 * 1024,
  maximumSubscriptions: 20,
  maximumChannelLength: 256,
  maximumBufferedBytes: 512 * 1024,
  messagesPerWindow: 30,
  messageWindowMs: 10_000,
  sessionValidationIntervalMs: 60_000,
});

Defaults are bounded; choose narrower values for the application's event size and connection profile when possible.

Attach sequence​

When a connection is attached:

  1. Origin is checked against the allowlist.
  2. The session token is validated.
  3. The connection is indexed by session hash.
  4. A ready message is sent.

Authentication failure closes the connection with 1008.

Channel subscription​

Generate channel names on the server or validate them with a narrow parser. Control characters and excessive lengths are rejected.

authorize: async ({ userId, channel }) => {
  const guildId = parseGuildChannel(channel);
  return guildId ? canUserViewGuild(userId, guildId) : false;
}

After asynchronous authorization completes, 1.1.0 rechecks that the connection is still active and has subscription capacity. Concurrent subscribe requests cannot bypass the limit.

WebSocket​

import { attachWebSocket } from "@kavtuai/guildgate/realtime";

await attachWebSocket({
  socket,
  hub,
  origin: request.headers.origin,
  sessionToken,
  eventLog,
  authorize,
});

WebSocket messages pass through hub.acceptMessage() first. Subscribe and unsubscribe are not applied before payload size, message rate, and JSON shape are validated.

Socket.IO parity​

import { attachSocketIo } from "@kavtuai/guildgate/realtime";

await attachSocketIo({
  socket,
  hub,
  eventLog,
  authorize,
});

In 1.1.0, subscribe, unsubscribe, and heartbeat events use the same controls as WebSocket messages. Socket.IO acknowledgements, replay, and backpressure derived from transport writable state are supported.

SSE​

createServerSentEventStream() uses a bounded event queue. When the writer reports backpressure, it waits for drain; when maximumBufferedEvents is exceeded, the stream closes.

SSE has no inbound message channel. The application should authorize before opening the stream and periodically revalidate long-lived access.

Sequence and replay​

createSequencedPublisher() generates per-channel sequences. A client can send its last sequence as afterSequence and receive a bounded replay from the event log.

Limit replay size. When a cursor is too old, provide a full state snapshot or reload path.

Session revocation​

hub.revokeSession(sessionIdHash) closes all connections for that session. In multi-worker deployments, replace MemorySessionRevocationBus with a shared pub/sub implementation.

A failed revocation listener must not stop other listeners. 1.1.0 isolates listener failures.

Backpressure and send failure​

If bufferedAmount exceeds the limit, the client closes with 1013. When the send promise fails, the connection record is removed so a broken connection does not remain in the hub.

Operations checklist​

  • origin and session are checked during handshake
  • channel authorization runs on the server
  • Socket.IO events pass through hub controls
  • connection lifetime and idle time are bounded
  • session is revalidated periodically
  • replay batches are bounded
  • slow clients are closed
  • hub.close() runs during shutdown
Last updated on