Building a WebSockets Server with Cloudflare Durable Objects

const http = require('http');

const server = http.createServer((req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  res.end('This is a WebSocket server\n');
});

server.on('upgrade', (req, socket, head) => {
  // verify the upgrade header
  if (req.headers['upgrade'] !== 'websocket') {
    socket.destroy();
    return;
  }

  const acceptKey = req.headers['sec-websocket-key'];

  // establish the WebSocket connection
  socket.write(responseHeaders.join('\r\n') + '\r\n\r\n');

  // receive and send messages
  socket.on('data', (buffer) => {
    const message = buffer.toString('utf8');
    console.log('Received:', message);
    const response = `Server says: ${message}`;
    socket.write(encodeMessage(response));
  });

  socket.on('end', () => {
    console.log('Client disconnected');
  });
});

function generateAcceptValue(acceptKey) {

function encodeMessage(message) {

server.listen(8080, () => {});

const socket = new WebSocket('ws://localhost:8080');
socket.onmessage = (event) => {
    console.log("Received message from server:", event.data);
};

socket.onopen = () => {
    socket.send('Hello from client');
};

const clients = []

server.on('upgrade', (req, socket, head) => {
  // code to upgrade header
  // ...

  clients.push(socket);

  socket.on('data', (buffer) => {
    clients.forEach((client) => {
      if (client.readyState === client.OPEN) {
        client.write(buffer);
      }
    });
  });

  // handle client disconnecting
  socket.on('end', () => {
    const index = clients.indexOf(socket);
    if (index !== -1) {
      clients.splice(index, 1);
    }
  });
});

pnpm create cloudflare --framework=hono

type Session = {
  name?: string;
};

export class Room extends DurableObject<Env> {
  private storage: DurableObjectStorage;

  private sessions: Map<WebSocket, Session>;
store connected sessions

  constructor(ctx: DurableObjectState, env: Env) {
    this.sessions = new Map();
  }

  async fetch(request: Request) {
    // accept socket ...

    await this.handleSession(server);

    // return response with client socket ...
  }

  async handleSession(ws: WebSocket) {
    const newSession: Session = { name: undefined };
register a new session whose username is unset
    this.sessions.set(ws, newSession);
  }

  async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    const session = this.sessions.get(ws);

    if (!session) {
      ws.close(1011, "Session not found");
      return;
    }

    const payload = JSON.parse(message as string) as any;
    if (payload.type === "join") {
handle join message
      session.name = payload.name;
      this.broadcast(`${session.name} joined the room | ${payload.timestamp}`);
    } else if (payload.type === "message") {
handle general chat message
      this.broadcast(
        `${session.name}: ${payload.message} | ${payload.timestamp}`
      );
    }
  }

  async webSocketClose(ws: WebSocket, code?: number, reason?: string) {
    this.sessions.delete(ws);
    ws.close(code || 1000, reason || "durable object closed websocket");
  }

  private broadcast(message: any) {
broadcast message to all connected users
    this.sessions.forEach((session, ws) => {
      if (session.name) {
        ws.send(JSON.stringify(message));
      }
    });
  }
}

constructor(ctx: DurableObjectState, env: Env) {
    // ...

    this.sessions = new Map();
    this.ctx.blockConcurrencyWhile(async () => {
        this.ctx.getWebSockets().forEach((websocket) => {
            this.sessions.set(websocket, {
                ...websocket.deserializeAttachment(),
restore session data
            });
        });
    });
    // ...
}

async handleSession(ws: WebSocket) {
    const newSession: Session = { name: undefined };

    // it's not necessary to serialize here because session data doesn't contain anything yet, but your app could be more complex and do have some state that needs to be saved.
    ws.serializeAttachment({
      ...session,
    });
    this.sessions.set(ws, newSession);
}

async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    // ...

    if (payload.type === "join") {
        session.name = payload.name;
        ws.serializeAttachment({
serialize session data to prepare for hibernation
            ...ws.deserializeAttachment(),
            name: payload.name,
        });
    }

  // ...
}

type Session = {
  name?: string;
  blockedMessages: string[];
};

constructor(ctx: DurableObjectState, env: Env) {
    // ...

    this.ctx.blockConcurrencyWhile(async () => {
        this.ctx.getWebSockets().forEach((websocket) => {
            this.sessions.set(websocket, {
                ...websocket.deserializeAttachment(),
                blockedMessages: [],
            });
        });
    });

    // ...
}

async handleSession(server: WebSocket, ip: string) {
    const newSession: Session = {
        name: undefined,
        blockedMessages: [],
    };

    server.serializeAttachment({
        ...session
    });

    this.sessions.set(server, session);

    const historyMessages = await this.storage.list({
retrieve previous messages
        reverse: true,
        limit: 100,
    });
    const backlog = [...historyMessages.values()];
    backlog.reverse();
    backlog.forEach((message) => {
        session.blockedMessages.push(message);
enqueue previous messages
    });
}

async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    // ...

    const payload = JSON.parse(message as string) as any;
    if (payload.type === "join") {
        session.name = payload.name;
        ws.serializeAttachment({
            ...ws.deserializeAttachment(),
            name: payload.name,
        });

        session.blockedMessages.forEach((message) => {
dequeue previous messages when username is received
            ws.send(JSON.stringify(message));
        });
        session.blockedMessages = [];

        this.broadcast(`${session.name} joined the room | ${payload.timestamp}`);

        return;
    }


    if (payload.type === "chat") {
        if (!session.name) {
            ws.close(1011, "Session without name");
            return;
        }

        this.broadcast(
            `${session.name}: ${payload.message} | ${payload.timestamp}`
        );
        const key = new Date(payload.timestamp).toISOString();
        await this.storage.put(key, JSON.stringify(data));
store chat message
    }
}