Build a WebSocket Gateway for Real-Time Features

// src/gateway/connection-manager.ts
import { Redis } from 'ioredis';
import { randomUUID } from 'crypto';

const redis = new Redis(process.env.REDIS_URL!);
const NODE_ID = randomUUID().slice(0, 8);

// In-memory connections for this node
const connections = new Map<string, {
  ws: WebSocket;
  userId: string;
  channels: Set<string>;
  connectedAt: number;
}>();

export function addConnection(connId: string, ws: WebSocket, userId: string): void {
  connections.set(connId, {
    ws,
    userId,
    channels: new Set(),
    connectedAt: Date.now(),
  });

  // Register in Redis for cross-node routing
  redis.pipeline()
    .hset(`ws:conn:${connId}`, 'userId', userId, 'node', NODE_ID, 'connectedAt', Date.now().toString())
    .sadd(`ws:user:${userId}`, connId)
    .expire(`ws:conn:${connId}`, 86400)
    .exec();
}

export function removeConnection(connId: string): void {
  const conn = connections.get(connId);
  if (!conn) return;

  // Unsubscribe from all channels
  for (const channel of conn.channels) {
    leaveChannel(connId, channel);
  }

  connections.delete(connId);

  redis.pipeline()
    .del(`ws:conn:${connId}`)
    .srem(`ws:user:${conn.userId}`, connId)
    .exec();
}

export async function joinChannel(connId: string, channel: string): Promise<void> {
  const conn = connections.get(connId);
  if (!conn) return;

  conn.channels.add(channel);
  await redis.sadd(`ws:channel:${channel}`, connId);
  await redis.sadd(`ws:channel:${channel}:nodes`, NODE_ID);

  // Notify presence
  await publishToChannel(channel, {
    type: 'presence.join',
    userId: conn.userId,
    channel,
  });
}

export async function leaveChannel(connId: string, channel: string): Promise<void> {
  const conn = connections.get(connId);
  if (!conn) return;

  conn.channels.delete(channel);
  await redis.srem(`ws:channel:${channel}`, connId);

  // Check if this node still has connections to this channel
  const localConns = [...connections.values()].filter(c => c.channels.has(channel));
  if (localConns.length === 0) {
    await redis.srem(`ws:channel:${channel}:nodes`, NODE_ID);
  }

  await publishToChannel(channel, {
    type: 'presence.leave',
    userId: conn.userId,
    channel,
  });
}

export async function publishToChannel(channel: string, message: any): Promise<void> {
  // Publish to Redis pub/sub for cross-node delivery
  await redis.publish(`ws:pubsub:${channel}`, JSON.stringify(message));
}

// Send to local connections subscribed to this channel
export function deliverToLocalChannel(channel: string, message: any): void {
  const payload = JSON.stringify(message);
  for (const [connId, conn] of connections) {
    if (conn.channels.has(channel) && conn.ws.readyState === 1) {
      conn.ws.send(payload);
    }
  }
}

export function sendToConnection(connId: string, message: any): void {
  const conn = connections.get(connId);
  if (conn && conn.ws.readyState === 1) {
    conn.ws.send(JSON.stringify(message));
  }
}

// Send to a specific user (all their connections)
export async function sendToUser(userId: string, message: any): Promise<void> {
  const connIds = await redis.smembers(`ws:user:${userId}`);
  const payload = JSON.stringify(message);

  for (const connId of connIds) {
    const conn = connections.get(connId);
    if (conn && conn.ws.readyState === 1) {
      conn.ws.send(payload);
    } else {
      // Connection on another node — publish via Redis
      await redis.publish(`ws:direct:${connId}`, payload);
    }
  }
}

// src/gateway/pubsub-bridge.ts
import { Redis } from 'ioredis';
import { deliverToLocalChannel, sendToConnection } from './connection-manager';

export async function startPubSubBridge(): Promise<void> {
  const sub = new Redis(process.env.REDIS_URL!);

  // Subscribe to channel messages
  sub.psubscribe('ws:pubsub:*');
  sub.psubscribe('ws:direct:*');

  sub.on('pmessage', (pattern, channel, message) => {
    const data = JSON.parse(message);

    if (channel.startsWith('ws:pubsub:')) {
      const wsChannel = channel.replace('ws:pubsub:', '');
      deliverToLocalChannel(wsChannel, data);
    } else if (channel.startsWith('ws:direct:')) {
      const connId = channel.replace('ws:direct:', '');
      sendToConnection(connId, data);
    }
  });
}

// src/gateway/presence.ts
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);

export async function getChannelPresence(channel: string): Promise<{
  count: number;
  users: Array<{ userId: string; connectedAt: number }>;
}> {
  const connIds = await redis.smembers(`ws:channel:${channel}`);
  const users: any[] = [];
  const seen = new Set<string>();

  for (const connId of connIds) {
    const data = await redis.hgetall(`ws:conn:${connId}`);
    if (data.userId && !seen.has(data.userId)) {
      seen.add(data.userId);
      users.push({ userId: data.userId, connectedAt: parseInt(data.connectedAt) });
    }
  }

  return { count: users.length, users };
}

// Typing indicator (ephemeral, not persisted)
export async function setTyping(channel: string, userId: string): Promise<void> {
  await redis.setex(`ws:typing:${channel}:${userId}`, 5, '1');
  // Broadcast to channel
  const { publishToChannel } = await import('./connection-manager');
  await publishToChannel(channel, { type: 'typing.start', userId, channel });
}