WebSockets

// The function half — runs once, at the handshake.
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.headers.get("Upgrade")?.toLowerCase() !== "websocket") {
      return new Response("expected websocket", { status: 426 });
    }
    const user = await authenticate(req);      // per-connection work — done once, here
    if (!user) return new Response(null, { status: 401 });

    const headers = new Headers(req.headers);
    headers.set("x-user", user.id);            // what you learned rides as headers
    return env.ROOM.idFromName(roomFor(req)).fetch(new Request(req, { headers }));
  },
};

// The actor half — owns the live socket from here on.
export class Room extends StatefulActor {
  async webSocket(ws: WebSocket, req: Request): Promise<void> {
    const user = req.headers.get("x-user");    // already authenticated
    ws.on("message", async (data) => {         // messages dispatch one at a time
      const n = ((await this.ctx.storage.get<number>("n")) ?? 0) + 1;
      await this.ctx.storage.put("n", n);
      ws.send(`${user} #${n}: ${data}`);       // streams straight back
    });
  }
}

async webSocket(ws: WebSocket, req: Request): Promise<void>

# telnyx.toml
name = "chat"
main = "src/index.ts"
compatibility_date = "2026-05-01"

[[actors]]
binding = "ROOM"
type    = "ChatRoom"

// src/index.ts
import { StatefulActor, type ActorNamespace } from "@telnyx/edge-runtime";
import type { WebSocket } from "ws";
// `ws` is a transitive dependency of @telnyx/edge-runtime. With npm it resolves
// automatically; with pnpm or other strict layouts, install @types/ws yourself.

export class ChatRoom extends StatefulActor {
  async webSocket(ws: WebSocket, req: Request): Promise<void> {
    // The front door authenticated this connection and stamped x-user; by the
    // time the socket reaches the actor, it is already authenticated.
    const user = req.headers.get("x-user") ?? "anonymous";

    ws.send(JSON.stringify({ type: "welcome", user, present: this.ctx.count() }));

    ws.on("message", async (data, isBinary) => {
      if (isBinary) return; // this room speaks text JSON

      // Parse defensively: an uncaught throw in a handler closes the
      // socket with code 1011.
      let text: string;
      try {
        ({ text } = JSON.parse(String(data)) as { text: string });
      } catch {
        ws.send(JSON.stringify({ type: "error", error: "frames must be JSON" }));
        return;
      }

      // Durable read-modify-write. Messages dispatch one at a time per
      // instance, so this never races another socket's frame — no lock, no
      // transaction.
      const seq = ((await this.ctx.storage.get<number>("seq")) ?? 0) + 1;
      await this.ctx.storage.put("seq", seq);
      await this.ctx.storage.put(`msg:${seq}`, { user, text, at: Date.now() });

      // Fan out to every live socket on THIS room, then ack the sender.
      const delivered = this.ctx.broadcast(
        JSON.stringify({ type: "message", seq, user, text }),
      );
      ws.send(JSON.stringify({ type: "sent", seq, delivered }));
    });

    // Close and error events go through the same single-threaded dispatch;
    // a storage write in a close handler still commits. The runtime removes the socket from
    // count()/broadcast() on its own — register these only for your own
    // bookkeeping.
    ws.on("close", async (code) => {
      await this.ctx.storage.put("lastLeave", { user, code, at: Date.now() });
    });
    // Keep an error listener so transport errors land in your logs instead of
    // vanishing; the runtime already tears the socket down on error.
    ws.on("error", (err) => console.error("ws error", err));
  }

  // Plain RPC on the same actor — same instance, same one-at-a-time dispatch.
  async messageCount(): Promise<number> {
    return (await this.ctx.storage.get<number>("seq")) ?? 0;
  }
}

interface Env {
  ROOM: ActorNamespace;
}

export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    const url = new URL(req.url);
    const match = url.pathname.match(/^\/rooms\/([^/]+)$/);
    if (!match) return new Response("not found", { status: 404 });

    if (req.headers.get("Upgrade")?.toLowerCase() !== "websocket") {
      return new Response("expected websocket", { status: 426 });
    }

    // Browser WebSocket clients cannot set request headers, so credentials
    // ride the URL. This code runs once per connection — the right place to
    // verify a signed token. The example only checks shape: forwarded header
    // values must be printable ASCII, and should be bounded.
    const user = url.searchParams.get("user");
    if (!user || !/^[\x20-\x7e]{1,128}$/.test(user)) {
      return new Response("missing or malformed ?user=", { status: 401 });
    }

    // Attach what the front door learned; overwrite anything client-supplied.
    const headers = new Headers(req.headers);
    headers.set("x-user", user);

    // Hand off: the actor named by the room owns the socket from here on.
    return env.ROOM.idFromName(match[1]!).fetch(new Request(req, { headers }));
  },
};

websocat "wss://<your-function-host>/rooms/lobby?user=alice"
{"type":"welcome","user":"alice","present":1}
{"text":"hello"}
{"type":"message","seq":1,"user":"alice","text":"hello"}
{"type":"sent","seq":1,"delivered":1}

export class Room extends StatefulActor {
  // Keyed by socket, not by user: one user may hold several sockets (two tabs).
  private peers = new Map<WebSocket, { user: string }>();

  async webSocket(ws: WebSocket, req: Request): Promise<void> {
    const user = req.headers.get("x-user") ?? "anonymous";
    this.peers.set(ws, { user });
    // Close events dispatch one at a time like everything else, so this
    // delete never races a message handler mid-iteration.
    ws.on("close", () => this.peers.delete(ws));

    ws.on("message", (data) => {
      // Everyone except the sender:
      for (const peer of this.peers.keys()) {
        if (peer !== ws) peer.send(`${user}: ${data}`);
      }
    });
  }

  // One named user — every socket they have open. Call it from a message
  // handler that routes a direct-message frame, or expose it as an RPC.
  sendTo(user: string, data: string): number {
    let sent = 0;
    for (const [peer, meta] of this.peers) {
      if (meta.user === user) {
        peer.send(data);
        sent++;
      }
    }
    return sent;
  }
}

export class Ticker extends StatefulActor {
  async webSocket(ws: WebSocket, req: Request): Promise<void> {
    // First socket arms the loop; later sockets just join the fan-out.
    if ((await this.ctx.storage.getAlarm()) === null) {
      await this.ctx.storage.setAlarm(Date.now() + 3_000);
    }
  }

  async alarm(): Promise<void> {
    this.ctx.broadcast(JSON.stringify({ type: "tick", at: Date.now() }));
    if (this.ctx.count() > 0) {
      await this.ctx.storage.setAlarm(Date.now() + 3_000); // re-arm while anyone listens
    }
  }
}