WebSockets at Scale: Real-Time Backend Architecture

snippet-10.go

func (c *Client) writePump(conn *websocket.Conn) {
    ticker := time.NewTicker(30 * time.Second)
    defer func() {
        ticker.Stop()
        conn.Close()
    }()

    conn.SetPongHandler(func(string) error {
        conn.SetReadDeadline(time.Now().Add(65 * time.Second))
        return nil
    })

    for {
        select {
        case message, ok := <-c.Send:
            conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
            if !ok {
                conn.WriteMessage(websocket.CloseMessage, []byte{})
                return
            }
            if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
                return
            }

        case <-ticker.C:
            conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
            if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        }
    }
}

snippet-2.sh

# Per-process file descriptor limits — add to /etc/security/limits.conf
* soft nofile 1000000
* hard nofile 1000000

snippet-3.go

package hub

import "sync"

type Client struct {
    ID   string
    Send chan []byte
    done chan struct{}
}

type Hub struct {
    clients    map[string]*Client
    broadcast  chan []byte
    register   chan *Client
    unregister chan *Client
    mu         sync.RWMutex
}

func NewHub() *Hub {
    return &Hub{
        clients:    make(map[string]*Client),
        broadcast:  make(chan []byte, 256),
        register:   make(chan *Client),
        unregister: make(chan *Client),
    }
}

func (h *Hub) Run() {
    for {
        select {
        case client := <-h.register:
            h.mu.Lock()
            h.clients[client.ID] = client
            h.mu.Unlock()

        case client := <-h.unregister:
            h.mu.Lock()
            if _, ok := h.clients[client.ID]; ok {
                delete(h.clients, client.ID)
                close(client.Send)
            }
            h.mu.Unlock()

        case message := <-h.broadcast:
            h.mu.RLock()
            for _, client := range h.clients {
                select {
                case client.Send <- message:
                default:
                    // Buffer full — client is too slow, disconnect it
                    close(client.Send)
                    delete(h.clients, client.ID)
                }
            }
            h.mu.RUnlock()
        }
    }
}

snippet-4.go

package hub

import (
    "hash/fnv"
)

type ShardedHub struct {
    shards []*Hub
    count  uint32
}

func NewShardedHub(shardCount int) *ShardedHub {
    shards := make([]*Hub, shardCount)
    for i := range shards {
        shards[i] = NewHub()
        go shards[i].Run()
    }
    return &ShardedHub{shards: shards, count: uint32(shardCount)}
}

func (s *ShardedHub) shard(clientID string) *Hub {
    h := fnv.New32a()
    h.Write([]byte(clientID))
    return s.shards[h.Sum32()%s.count]
}

func (s *ShardedHub) Register(c *Client)   { s.shard(c.ID).register <- c }
func (s *ShardedHub) Unregister(c *Client) { s.shard(c.ID).unregister <- c }
func (s *ShardedHub) Send(clientID string, msg []byte) {
    s.shard(clientID).broadcast <- msg
}

snippet-5.go

package pubsub

import (
    "context"
    "github.com/redis/go-redis/v9"
)

type RedisBroker struct {
    client *redis.Client
    hub    *ShardedHub
}

func (b *RedisBroker) Subscribe(ctx context.Context, channel string) {
    sub := b.client.Subscribe(ctx, channel)
    ch := sub.Channel()

    go func() {
        defer sub.Close()
        for {
            select {
            case msg := <-ch:
                // Fan out to all local connections in this topic
                b.hub.BroadcastToTopic(channel, []byte(msg.Payload))
            case <-ctx.Done():
                return
            }
        }
    }()
}

func (b *RedisBroker) Publish(ctx context.Context, channel string, payload []byte) error {
    return b.client.Publish(ctx, channel, payload).Err()
}

snippet-6.sql

-- Track which server node holds each connection
CREATE TABLE connection_registry (
    connection_id   UUID PRIMARY KEY,
    server_node_id  TEXT NOT NULL,
    user_id         BIGINT,
    connected_at    TIMESTAMPTZ DEFAULT NOW(),
    last_seen_at    TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_connection_registry_user ON connection_registry(user_id);
CREATE INDEX idx_connection_registry_node ON connection_registry(server_node_id);

-- Periodically clean stale records from crashed nodes
DELETE FROM connection_registry
WHERE last_seen_at < NOW() - INTERVAL '2 minutes';

snippet-7.conf

upstream websocket_backend {
    ip_hash;  # sticky sessions via client IP
    server ws1.internal:8080;
    server ws2.internal:8080;
    server ws3.internal:8080;
    keepalive 1000;
}

server {
    listen 443 ssl;

    location /ws {
        proxy_pass http://websocket_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 3600s;   # 1 hour — match your heartbeat interval
        proxy_send_timeout 3600s;
    }
}

snippet-8.dockerfile

FROM golang:1.22-alpine AS builder
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-w -s" -o wsserver ./cmd/server

FROM gcr.io/distroless/static-debian12
COPY --from=builder /app/wsserver /wsserver
EXPOSE 8080
ENTRYPOINT ["/wsserver"]

snippet-9.yaml

# kubernetes deployment excerpt
spec:
  terminationGracePeriodSeconds: 60
  containers:
    - name: wsserver
      image: your-registry/wsserver:latest
      ports:
        - containerPort: 8080
      lifecycle:
        preStop:
          exec:
            command: ["/bin/sh", "-c", "sleep 5"]  # drain LB before SIGTERM
      resources:
        limits:
          memory: "2Gi"
        requests:
          memory: "1Gi"
          cpu: "500m"

snippet.sh

# /etc/sysctl.conf — apply with: sysctl -p
fs.file-max = 10000000
net.core.somaxconn = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.ip_local_port_range = 1024 65535
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 15
net.core.rmem_max = 16777216
net.core.wmem_max = 16777216