godot-webrtc-mplayer-testing/server.ts

import { randomId, randomSecret } from "./gen.ts";

const MAX_PEERS = 4096;
const MAX_LOBBIES = 1024;
const PORT = 80;

const NO_LOBBY_TIMEOUT = 10000;
const SEAL_CLOSE_TIMEOUT = 10000;
// const PING_INTERVAL = 10000;

const STR_NO_LOBBY = "Have not joined lobby yet";
const STR_HOST_DISCONNECTED = "Room host has disconnected";
const STR_ONLY_HOST_CAN_SEAL = "Only host can seal the lobby";
const STR_SEAL_COMPLETE = "Seal complete";
const STR_TOO_MANY_LOBBIES = "Too many lobbies open, disconnecting";
const STR_ALREADY_IN_LOBBY = "Already in a lobby";
const STR_LOBBY_DOES_NOT_EXIST = "Lobby does not exist";
const STR_LOBBY_IS_SEALED = "Lobby is sealed";
const STR_INVALID_FORMAT = "Invalid message format";
const STR_NEED_LOBBY = "Invalid message when not in a lobby";
const STR_SERVER_ERROR = "Server error, lobby not found";
const STR_INVALID_DEST = "Invalid destination";
const STR_INVALID_CMD = "Invalid command";
const STR_TOO_MANY_PEERS = "Too many peers connected";

// TODO: setup regular pings?

// state
const lobbies = new Map();
let peersCount = 0;

class ProtoError extends Error {
  code: number;

  constructor(code: number, message: string) {
    super(message);
    this.code = code;
  }
}

class Peer {
  id: number;
  ws: WebSocket;
  lobby: string;
  timeout: number;

  constructor(id: number, ws: WebSocket) {
    this.id = id;
    this.ws = ws;
    this.lobby = "";

    // close connection after 10 sec if client has not joined a lobby
    this.timeout = setTimeout(() => {
      if (!this.lobby) ws.close(4000, STR_NO_LOBBY);
    }, NO_LOBBY_TIMEOUT);
  }

  joinLobby(lobbyName: string) {
    if (lobbyName === "") {
      if (lobbies.size >= MAX_LOBBIES) {
        throw new ProtoError(4000, STR_TOO_MANY_LOBBIES);
      }
      // Peer must not already be in a lobby
      if (this.lobby !== "") {
        throw new ProtoError(4000, STR_ALREADY_IN_LOBBY);
      }
      lobbyName = randomSecret();
      lobbies.set(lobbyName, new Lobby(lobbyName, this.id));
      console.log(`Peer ${this.id} created lobby ${lobbyName}`);
      console.log(`Open lobbies: ${lobbies.size}`);
    }
    const lobby = lobbies.get(lobbyName);
    if (!lobby) throw new ProtoError(4000, STR_LOBBY_DOES_NOT_EXIST);
    if (lobby.sealed) throw new ProtoError(4000, STR_LOBBY_IS_SEALED);
    this.lobby = lobbyName;
    console.log(
      `Peer ${this.id} joining lobby ${lobbyName} ` +
        `with ${lobby.peers.length} peers`,
    );
    lobby.join(this);
    this.ws.send(`J: ${lobbyName}\n`);
  }
}

class Lobby {
  name: string;
  host: number;
  peers: Peer[];
  sealed: boolean;
  closeTimer: number;

  constructor(name: string, host: number) {
    this.name = name;
    this.host = host;
    this.peers = [];
    this.sealed = false;
    this.closeTimer = -1;
  }

  getPeerId(peer: Peer) {
    if (this.host === peer.id) return 1;
    return peer.id;
  }

  join(peer: Peer) {
    const assigned = this.getPeerId(peer);
    peer.ws.send(`I: ${assigned}\n`);
    this.peers.forEach((p) => {
      p.ws.send(`N: ${assigned}\n`);
      peer.ws.send(`N: ${this.getPeerId(p)}\n`);
    });
    this.peers.push(peer);
  }

  leave(peer: Peer) {
    const idx = this.peers.findIndex((p) => peer === p);
    if (idx === -1) return false;
    const assigned = this.getPeerId(peer);
    const close = assigned === 1;
    this.peers.forEach((p) => {
      try {
        // room host disconnected
        if (close) p.ws.close(4000, STR_HOST_DISCONNECTED);
        // notify peers
        else p.ws.send(`D: ${assigned}\n`);
      } catch (e) {
        console.error(`Error when leaving: ${e}`);
      }
    });
    this.peers.splice(idx, 1);
    if (close && this.closeTimer >= 0) {
      // we are closing already.
      clearTimeout(this.closeTimer);
      this.closeTimer = -1;
    }
    return close;
  }

  seal(peer: Peer) {
    // only host can seal
    if (peer.id !== this.host) {
      throw new ProtoError(4000, STR_ONLY_HOST_CAN_SEAL);
    }
    this.sealed = true;
    this.peers.forEach((p) => {
      p.ws.send("S: \n");
    });
    console.log(
      `Peer ${peer.id} sealed lobby ${this.name} ` +
        `with ${this.peers.length} peers`,
    );
    this.closeTimer = setTimeout(() => {
      // close peer connection to host (and thus the lobby)
      this.peers.forEach((p) => {
        p.ws.close(1000, STR_SEAL_COMPLETE);
      });
    }, SEAL_CLOSE_TIMEOUT);
  }
}

function parseMsg(peer: Peer, msg: string) {
  const sep = msg.indexOf("\n");
  if (sep < 0) throw new ProtoError(4000, STR_INVALID_FORMAT);

  const cmd = msg.slice(0, sep);
  if (cmd.length < 3) throw new ProtoError(4000, STR_INVALID_FORMAT);

  const data = msg.slice(sep);

  // join
  if (cmd.startsWith("J: ")) {
    peer.joinLobby(cmd.substr(3).trim());
    return;
  }

  if (!peer.lobby) throw new ProtoError(4000, STR_NEED_LOBBY);
  const lobby = lobbies.get(peer.lobby);
  if (!lobby) throw new ProtoError(4000, STR_SERVER_ERROR);

  // seal
  if (cmd.startsWith("S: ")) {
    lobby.seal(peer);
    return;
  }

  // Message relaying format:
  //
  // [O|A|C]: DEST_ID\n
  // PAYLOAD
  //
  // O: Client is sending an offer.
  // A: Client is sending an answer.
  // C: Client is sending a candidate.
  let destId = parseInt(cmd.substr(3).trim());
  // Dest is not an ID.
  if (!destId) throw new ProtoError(4000, STR_INVALID_DEST);
  if (destId === 1) destId = lobby.host;
  const dest = lobby.peers.find((p: Peer) => p.id === destId);
  // Dest is not in this room.
  if (!dest) throw new ProtoError(4000, STR_INVALID_DEST);

  function isCmd(what: string) {
    return cmd.startsWith(`${what}: `);
  }
  if (isCmd("O") || isCmd("A") || isCmd("C")) {
    dest.ws.send(cmd[0] + ": " + lobby.getPeerId(peer) + data);
    return;
  }
  throw new ProtoError(4000, STR_INVALID_CMD);
}

console.log(`Server running on port ${PORT}`);
const server = Deno.listen({ port: PORT });
for await (const conn of server) {
  (async () => {
    const httpConn = Deno.serveHttp(conn);
    for await (const requestEvent of httpConn) {
      if (requestEvent) {
        try {
          const { socket, response } = Deno.upgradeWebSocket(
            requestEvent.request,
          );
          const id = randomId();
          const peer = new Peer(id, socket);
          socket.onopen = (_ev) => {
            if (peersCount >= MAX_PEERS) {
              socket.close(4000, STR_TOO_MANY_PEERS);
              return;
            }
            peersCount++;
          };
          socket.onmessage = (ev) => {
            try {
              parseMsg(peer, ev.data);
            } catch (e) {
              const code = e.code || 4000;
              console.log(`Error parsing message from ${id}:\n` + ev.data);
              socket.close(code, e.message);
            }
          };
          socket.onclose = (ev) => {
            peersCount--;
            console.log(
              `Connection with peer ${peer.id} closed ` +
                `with reason: ${ev.reason}`,
            );
            if (
              peer.lobby &&
              lobbies.has(peer.lobby) &&
              lobbies.get(peer.lobby).leave(peer)
            ) {
              lobbies.delete(peer.lobby);
              console.log(`Deleted lobby ${peer.lobby}`);
              console.log(`Open lobbies: ${lobbies.size}`);
              peer.lobby = "";
            }
            if (peer.timeout >= 0) {
              clearTimeout(peer.timeout);
              peer.timeout = -1;
            }
          };
          socket.onerror = (e) => console.error("WebSocket error:", e);
          requestEvent.respondWith(response);
        } catch (e) {
          console.log(`Error during connection:`, e);
        }
      }
    }
  })();
}