WIP: Horizontal Scaling

.gitignore

@@ -1,2 +1,4 @@
 dist/
-node_modules/
+node_modules/
+.idea/
+*.iml

package.json

@@ -4,22 +4,30 @@
  "main": "index.js",
  "license": "GPL-3.0-or-later",
  "dependencies": {
+  "amqplib": "^0.6.0",
   "express": "^4.17.1",
   "morgan": "^1.10.0",
   "public-ip": "^4.0.2",
   "pug": "^3.0.0",
+  "redis": "^3.0.2",
+  "redlock": "^4.2.0",
   "socket.io": "^2.3.0",
-  "tracer": "^1.1.4"
+  "tracer": "^1.1.4",
+  "uuid": "^8.3.2"
  },
  "scripts": {
   "start": "yarn compile && node dist/index.js",
   "compile": "tsc",
   "postinstall": "yarn compile"
  },
  "devDependencies": {
-  "@types/express": "^4.17.8",
+  "@types/amqplib": "^0.5.16",
+  "@types/express": "^4.17.9",
   "@types/morgan": "^1.9.2",
+  "@types/public-ip": "^3.1.0",
+  "@types/redis": "^2.8.28",
   "@types/socket.io": "^2.1.11",
+  "@types/uuid": "^8.3.0",
   "typescript": "^4.0.5"
  }
 }

src/cache/Cache.ts

@@ -0,0 +1,20 @@
+import {Player} from "../model/Player";
+
+export interface Cache {
+  /**
+   * Retrieve room players
+   * @param roomId The roomID to retrieve players from
+   */
+  retrieveRoomPlayers(roomId: string): Promise<Array<Player>>;
+
+  /**
+   * Add a player to a room
+   * @param roomId The room to add the player in
+   * @param player The player's data
+   */
+  addPlayerToRoom(roomId: string, player: Player): Promise<void>;
+  removePlayerFromRoom(roomId: string, playerId: string): Promise<void>;
+  updateSession(socketId: string, serverId: string): Promise<void>;
+  clearSession(socketId: string): Promise<void>;
+  getSession(socketId: string): Promise<string>;
+}

src/cache/LocalCache.ts

@@ -0,0 +1,47 @@
+import {Cache} from "./Cache";
+import {Player} from "../model/Player";
+
+export class LocalCache implements Cache {
+  private rooms: Map<string, Array<Player>> = new Map();
+
+  async addPlayerToRoom(roomId: string, player: Player): Promise<void> {
+    let room = this.rooms.get(roomId);
+    if (!room) {
+      room = [];
+      this.rooms.set(roomId, room);
+    }
+    room.push(player);
+  }
+
+  async clearSession(socketId: string): Promise<void> {
+    return Promise.resolve(undefined);
+  }
+
+  async getSession(socketId: string): Promise<string> {
+    return Promise.resolve("");
+  }
+
+  async removePlayerFromRoom(roomId: string, playerId: string): Promise<void> {
+    const room = this.rooms.get(roomId);
+    if (room) {
+      let i = room.length;
+      while (--i) {
+        if (room[i].id == playerId) {
+          room.splice(i, 1);
+        }
+      }
+      if (room.length == 0) {
+        this.rooms.delete(roomId);
+      }
+    }
+  }
+
+  async retrieveRoomPlayers(roomId: string): Promise<Array<Player>> {
+    return this.rooms.get(roomId) || [];
+  }
+
+  async updateSession(socketId: string, serverId: string): Promise<void> {
+    return Promise.resolve(undefined);
+  }
+
+}

src/cache/RedisCache.ts

@@ -0,0 +1,83 @@
+import {Cache} from "./Cache";
+import {Player} from "../model/Player";
+import Redis, {RedisClient} from "redis";
+
+export class RedisCache implements Cache {
+  private readonly ROOM_KEY_PRE = process.env.REDIS_ROOM_KEY_PRE || "crewlink:room:";
+  private readonly SESSION_KEY_PRE = process.env.REDIS_ROOM_KEY_PRE || "crewlink:session:";
+  private readonly redis: RedisClient;
+
+  constructor(redisUrl: string) {
+    this.redis = Redis.createClient(redisUrl);
+  }
+
+  async addPlayerToRoom(roomId: string, player: Player): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      this.redis.hset(`${this.ROOM_KEY_PRE}{${roomId}}`,
+        player.id, JSON.stringify(player), (err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve();
+      });
+    });
+  }
+
+  async removePlayerFromRoom(roomId: string, playerId: string): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      this.redis.hdel(`${this.ROOM_KEY_PRE}{${roomId}}`, playerId, (err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve();
+      });
+    });
+  }
+
+  async retrieveRoomPlayers(roomId: string): Promise<Array<Player>> {
+    return new Promise<Array<Player>>((resolve, reject) => {
+      this.redis.hgetall(`${this.ROOM_KEY_PRE}{${roomId}}`, (err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve(Object.keys(reply).map(playerId => JSON.parse(reply[playerId])));
+      })
+    });
+  }
+
+  clearSession(socketId: string): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      this.redis.del(`${this.SESSION_KEY_PRE}{${socketId}}`, (err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve();
+      });
+    });
+  }
+
+  updateSession(socketId: string, serverId: string): Promise<void> {
+    // FIXME: In a ideal world, a ping-pong mechanism must exist between client and server and this
+    //  should have a TTL and the expire value of this TTL should be updated on each ping, to avoid memory leaks
+    //  on redis
+    return new Promise<void>((resolve, reject) => {
+      this.redis.set(`${this.SESSION_KEY_PRE}{${socketId}}`, serverId,(err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve();
+      });
+    });
+  }
+
+  getSession(socketId: string): Promise<string | null> {
+    return new Promise<string>((resolve, reject) => {
+      this.redis.get(`${this.SESSION_KEY_PRE}{${socketId}}`,(err, reply) => {
+        if (err) {
+          reject(err);
+        }
+        resolve(reply);
+      });
+    });
+  }
+}