Build Real-Time Collaboration Cursors

npm install hono @hono/node-server @hono/node-ws ioredis zod
npm install -D typescript @types/node
npm pkg set type=module

// server.ts
import { Hono } from "hono";
import { serve } from "@hono/node-server";
import { createNodeWebSocket } from "@hono/node-ws";
import type { WSContext } from "hono/ws";
import { Redis } from "ioredis";
import { z } from "zod";

const redis = new Redis(process.env.REDIS_URL!); // commands + publishing
const sub = new Redis(process.env.REDIS_URL!);   // a subscribed connection can't run commands

const PALETTE = ["#EF4444", "#3B82F6", "#10B981", "#F59E0B", "#8B5CF6", "#EC4899", "#06B6D4", "#84CC16"];
const PRESENCE_TTL = 30; // seconds; the client heartbeat renews it

// What a client may send. Identity never comes from the message, only from the session.
const ClientMsg = z.discriminatedUnion("type", [
  z.object({ type: z.literal("cursor"), x: z.number().finite(), y: z.number().finite(), elementId: z.string().max(64).optional() }),
  z.object({ type: z.literal("selection"), ids: z.array(z.string().max(64)).max(200) }),
  z.object({ type: z.literal("heartbeat"), idle: z.boolean().default(false) }),
]);

type Peer = { ws: WSContext; userId: string };
const peers = new Map<string, Set<Peer>>(); // docId -> sockets connected to THIS server

// Every server instance subscribes once and relays to its own sockets.
await sub.psubscribe("collab:*");
sub.on("pmessage", (_pattern, channel, raw) => {
  const docId = channel.slice("collab:".length);
  const { from, payload } = JSON.parse(raw) as { from: string; payload: unknown };
  for (const p of peers.get(docId) ?? []) if (p.userId !== from) p.ws.send(JSON.stringify(payload));
});
const publish = (docId: string, from: string, payload: object) =>
  redis.publish(`collab:${docId}`, JSON.stringify({ from, payload }));

const presenceKey = (docId: string, userId: string) => `presence:${docId}:${userId}`;

// Members whose last heartbeat is inside the TTL window
async function members(docId: string) {
  await redis.zremrangebyscore(`room:${docId}`, "-inf", Date.now() - PRESENCE_TTL * 1000); // drop crashed tabs
  const ids = await redis.zrangebyscore(`room:${docId}`, Date.now() - PRESENCE_TTL * 1000, "+inf");
  if (!ids.length) return [];
  const rows = await redis.mget(ids.map((id) => presenceKey(docId, id)));
  return rows.filter((r): r is string => r !== null).map((r) => JSON.parse(r));
}

const app = new Hono();
const { upgradeWebSocket, injectWebSocket } = createNodeWebSocket({ app });

app.get("/ws/:docId", upgradeWebSocket(async (c) => {
  const docId = c.req.param("docId") as string;
  // Browsers can't set headers on a WebSocket, so the token rides in the query string.
  // Use a short-lived, single-purpose token and keep query strings out of access logs.
  const session = await redis.get(`session:${c.req.query("token") ?? ""}`);
  const user = session ? (JSON.parse(session) as { userId: string; name: string }) : null;
  let peer: Peer | undefined;

  return {
    async onOpen(_evt, ws) {
      if (!user) return ws.close(4401, "unauthorized");
      peer = { ws, userId: user.userId };
      if (!peers.has(docId)) peers.set(docId, new Set());
      peers.get(docId)!.add(peer);

      const existing = await members(docId);
      const color = PALETTE[existing.length % PALETTE.length];
      const me = { userId: user.userId, name: user.name, color, selection: [] as string[], idle: false };
      await redis.set(presenceKey(docId, user.userId), JSON.stringify(me), "EX", PRESENCE_TTL);
      await redis.zadd(`room:${docId}`, Date.now(), user.userId);
      ws.send(JSON.stringify({ type: "welcome", me, others: existing.filter((m) => m.userId !== user.userId) }));
      await publish(docId, user.userId, { type: "joined", user: me });
    },

    async onMessage(evt, _ws) {
      if (!user) return;
      let json: unknown;
      try { json = JSON.parse(String(evt.data)); } catch { return; } // not JSON: drop it
      const parsed = ClientMsg.safeParse(json);
      if (!parsed.success) return;
      const msg = parsed.data;
      const key = presenceKey(docId, user.userId);
      const stored = await redis.get(key);
      if (!stored) return;
      const me = JSON.parse(stored);

      if (msg.type === "cursor") {
        // Cursor positions are ephemeral: relay only, never store them.
        return void publish(docId, user.userId, { type: "cursor", userId: user.userId, x: msg.x, y: msg.y, elementId: msg.elementId });
      }
      if (msg.type === "selection") me.selection = msg.ids;
      if (msg.type === "heartbeat") me.idle = msg.idle;
      await redis.set(key, JSON.stringify(me), "EX", PRESENCE_TTL);
      await redis.zadd(`room:${docId}`, Date.now(), user.userId);
      if (msg.type === "selection") await publish(docId, user.userId, { type: "selection", userId: user.userId, ids: msg.ids });
      if (msg.type === "heartbeat") await publish(docId, user.userId, { type: "idle", userId: user.userId, idle: msg.idle });
    },

    async onClose() {
      if (!user || !peer) return;
      peers.get(docId)?.delete(peer);
      await redis.del(presenceKey(docId, user.userId));
      await redis.zrem(`room:${docId}`, user.userId);
      await publish(docId, user.userId, { type: "left", userId: user.userId });
    },
  };
}));

// Bind to loopback; put your reverse proxy (TLS) in front for public traffic.
injectWebSocket(serve({ fetch: app.fetch, port: Number(process.env.PORT ?? 8787), hostname: "127.0.0.1" }));

const ws = new WebSocket(`wss://app.example.com/ws/${docId}?token=${wsToken}`);
let last = 0;
canvas.addEventListener("pointermove", (e) => {
  const now = performance.now();
  if (now - last < 33 || ws.readyState !== WebSocket.OPEN) return;
  last = now;
  ws.send(JSON.stringify({ type: "cursor", x: e.offsetX, y: e.offsetY }));
});
// Renew presence every 10 s; set idle after 60 s without input
setInterval(() => ws.send(JSON.stringify({ type: "heartbeat", idle: Date.now() - lastInput > 60_000 })), 10_000);