Merge pull request #18 from 5stackgg/main

Author: daniMilio

src/auth/auth.controller.ts

@@ -5,12 +5,15 @@ import { HasuraAction } from "../hasura/hasura.controller";
 import { DiscordGuard } from "./strategies/DiscordGuard";
 import { CacheService } from "../cache/cache.service";
 import { HasuraService } from "../hasura/hasura.service";
+import { SocketsGateway } from "src/sockets/sockets.gateway";
+import { RedisManagerService } from "src/redis/redis-manager/redis-manager.service";
 
 @Controller("auth")
 export class AuthController {
   constructor(
     private readonly cache: CacheService,
     private readonly hasura: HasuraService,
+    private readonly redis: RedisManagerService,
   ) {}
 
   @UseGuards(SteamGuard)
@@ -73,6 +76,10 @@ export class AuthController {
   @HasuraAction()
   public async logout(@Req() request: Request) {
     if (request.session) {
+      await this.redis
+        .getConnection()
+        .del(SocketsGateway.GET_PLAYER_CLIENT_LATENCY_TEST(request.session.id));
+
       request.session.destroy((err) => {
         if (err) {
           console.error("Error destroying session:", err);

src/auth/auth.module.ts

@@ -7,6 +7,7 @@ import { SteamSerializer } from "./strategies/SteamSerializer";
 import { DiscordStrategy } from "./strategies/DiscordStrategy";
 import { loggerFactory } from "../utilities/LoggerFactory";
 import { CacheModule } from "../cache/cache.module";
+import { RedisModule } from "../redis/redis.module";
 
 @Module({
   imports: [
@@ -15,6 +16,7 @@ import { CacheModule } from "../cache/cache.module";
     }),
     HasuraModule,
     CacheModule,
+    RedisModule,
   ],
   providers: [SteamStrategy, DiscordStrategy, SteamSerializer, loggerFactory()],
   controllers: [AuthController],

src/matchmaking/jobs/MarkPlayerOffline.ts

@@ -0,0 +1,50 @@
+import { Job } from "bullmq";
+import { WorkerHost } from "@nestjs/bullmq";
+import { MatchmakingQueues } from "../enums/MatchmakingQueues";
+import { UseQueue } from "../../utilities/QueueProcessors";
+import { MatchmakingLobbyService } from "../matchmaking-lobby.service";
+import { HasuraService } from "../../hasura/hasura.service";
+
+@UseQueue("Matchmaking", MatchmakingQueues.Matchmaking)
+export class MarkPlayerOffline extends WorkerHost {
+  constructor(
+    private readonly hasura: HasuraService,
+    private readonly matchmakingLobbyService: MatchmakingLobbyService,
+  ) {
+    super();
+  }
+
+  async process(
+    job: Job<{
+      steamId: string;
+    }>,
+  ): Promise<void> {
+    console.info(`Leaving queue for ${job.data.steamId}`);
+    const { steamId } = job.data;
+
+    await this.hasura.mutation({
+      delete_lobby_players: {
+        __args: {
+          where: {
+            steam_id: {
+              _eq: steamId,
+            },
+            status: {
+              _eq: "Accepted",
+            },
+          },
+        },
+        __typename: true,
+      },
+    });
+
+    const lobby = await this.matchmakingLobbyService.getPlayerLobby(steamId);
+
+    if (!lobby) {
+      return;
+    }
+
+    await this.matchmakingLobbyService.removeLobbyFromQueue(lobby.id);
+    await this.matchmakingLobbyService.removeLobbyDetails(lobby.id);
+  }
+}

src/matchmaking/matchmake.service.ts

@@ -412,7 +412,22 @@ export class MatchmakeService {
     await this.redis.expire(lockKey, seconds);
   }
 
-  // TODO - watch for a player leaving a lobby and force the entire lobby to be left the queue with an error
+  public async markOffline(steamId: string) {
+    await this.queue.add(
+      "MarkPlayerOffline",
+      {
+        steamId,
+      },
+      {
+        delay: 60 * 1000,
+        jobId: `matchmaking:mark-offline:${steamId}`,
+      },
+    );
+  }
+
+  public async cancelOffline(steamId: string) {
+    await this.queue.remove(`matchmaking:mark-offline:${steamId}`);
+  }
 
   private async createMatchConfirmation(
     region: string,

src/matchmaking/matchmaking-lobby.service.ts

@@ -31,8 +31,8 @@ export class MatchmakingLobbyService {
     this.redis = this.redisManager.getConnection();
   }
 
-  public async getPlayerLobby(user: User): Promise<PlayerLobby> {
-    let lobbyId = await this.getCurrentLobbyId(user.steam_id);
+  public async getPlayerLobby(steamId: string): Promise<PlayerLobby> {
+    let lobbyId = await this.getCurrentLobbyId(steamId);
 
     let lobby;
     if (validateUUID(lobbyId)) {
@@ -64,11 +64,11 @@ export class MatchmakingLobbyService {
     }
 
     if (!lobby) {
-      lobbyId = user.steam_id;
+      lobbyId = steamId;
       const { players_by_pk } = await this.hasura.query({
         players_by_pk: {
           __args: {
-            steam_id: user.steam_id,
+            steam_id: steamId,
           },
           name: true,
           steam_id: true,
@@ -271,8 +271,8 @@ export class MatchmakingLobbyService {
   }
 
   // TODO - extermly inefficient
-  public async sendQueueDetailsToPlayer(user: User) {
-    const lobby = await this.getPlayerLobby(user);
+  public async sendQueueDetailsToPlayer(steamId: string) {
+    const lobby = await this.getPlayerLobby(steamId);
 
     if (!lobby) {
       return;

src/matchmaking/matchmaking.gateway.ts

@@ -15,6 +15,7 @@ import { JoinQueueError } from "./utilities/joinQueueError";
 import { HasuraService } from "src/hasura/hasura.service";
 import { isRoleAbove } from "src/utilities/isRoleAbove";
 import { e_player_roles_enum } from "generated";
+import { SocketsGateway } from "src/sockets/sockets.gateway";
 
 @WebSocketGateway({
   path: "/ws/web",
@@ -88,18 +89,59 @@ export class MatchmakingGateway {
 
     let lobby;
     const user = client.user;
-    const { type, regions } = data;
 
     if (!user) {
       return;
     }
 
+    const latencyResults = await this.getLatencyResults(client);
+
+    const { server_regions } = await this.hasura.query({
+      server_regions: {
+        __args: {
+          where: {
+            status: {
+              _eq: "Online",
+            },
+          },
+        },
+        value: true,
+        is_lan: true,
+        status: true,
+      },
+    });
+
+    // TODO - rather adding all regions at once we should add them when expanding the search
+
+    let regions = [];
+    for (const region of data.regions) {
+      const server_region = server_regions.find((server_region) => {
+        return server_region.value === region;
+      });
+
+      if (!server_region) {
+        continue;
+      }
+
+      const latency =
+        latencyResults[region.toLocaleLowerCase().replace(" ", "_")];
+      if (!server_region.is_lan || latency?.isLan === true) {
+        regions.push(server_region.value);
+      }
+    }
+
+    if (regions.length === 0) {
+      throw new JoinQueueError("No regions available");
+    }
+
+    const { type } = data;
+
     try {
       if (!type || !regions || regions.length === 0) {
         throw new JoinQueueError("Missing Type or Regions");
       }
 
-      lobby = await this.matchmakingLobbyService.getPlayerLobby(user);
+      lobby = await this.matchmakingLobbyService.getPlayerLobby(user.steam_id);
 
       if (!lobby) {
         throw new JoinQueueError("Unable to find Player Lobby");
@@ -161,7 +203,9 @@ export class MatchmakingGateway {
       return;
     }
 
-    const lobby = await this.matchmakingLobbyService.getPlayerLobby(user);
+    const lobby = await this.matchmakingLobbyService.getPlayerLobby(
+      user.steam_id,
+    );
 
     if (!lobby) {
       return;
@@ -190,4 +234,24 @@ export class MatchmakingGateway {
       user.steam_id,
     );
   }
+
+  private async getLatencyResults(client: FiveStackWebSocketClient) {
+    const data = await this.redis.hgetall(
+      SocketsGateway.GET_PLAYER_CLIENT_LATENCY_TEST(client.sessionId),
+    );
+
+    const latencyResults: Record<
+      string,
+      {
+        isLan: boolean;
+        latency: number;
+      }
+    > = {};
+
+    for (const key in data) {
+      latencyResults[key] = JSON.parse(data[key]);
+    }
+
+    return latencyResults;
+  }
 }

src/matchmaking/matchmaking.module.ts

@@ -13,6 +13,7 @@ import { getQueuesProcessors } from "src/utilities/QueueProcessors";
 import { MatchmakingQueues } from "./enums/MatchmakingQueues";
 import { CancelMatchMaking } from "./jobs/CancelMatchMaking";
 import { MatchmakingController } from "./matchmaking.controller";
+import { MarkPlayerOffline } from "./jobs/MarkPlayerOffline";
 
 @Module({
   imports: [
@@ -33,6 +34,7 @@ import { MatchmakingController } from "./matchmaking.controller";
     MatchmakeService,
     MatchmakingLobbyService,
     CancelMatchMaking,
+    MarkPlayerOffline,
     ...getQueuesProcessors("Matchmaking"),
     loggerFactory(),
   ],

src/signal-server/signal-server.gateway.ts

@@ -13,6 +13,7 @@ import { RegionSignalData } from "./types/SignalData";
 
 interface WebRTCClient extends WebSocket {
   id: string;
+  sessionId: string;
 }
 
 @WebSocketGateway({
@@ -54,6 +55,7 @@ export class SignalServerGateway {
           signal,
           peerId,
           clientId: client.id,
+          sessionId: client.sessionId,
         },
       }),
     );

src/sockets/sockets.gateway.ts

@@ -27,14 +27,6 @@ export class SocketsGateway {
   private nodeId: string = process.env.POD_NAME;
   private clients: Map<string, FiveStackWebSocketClient> = new Map();
 
-  public static GET_AVAILABLE_NODES_KEY() {
-    return `available-socket-nodes`;
-  }
-
-  public static GET_NODE_STATUS_KEY(nodeId: string) {
-    return `socket-node:${nodeId}:status`;
-  }
-
   public static GET_PLAYER_KEY(steamId: string) {
     return `players:${steamId}`;
   }
@@ -55,6 +47,10 @@ export class SocketsGateway {
     return `${SocketsGateway.GET_PLAYER_CLIENTS_BY_NODE(steamId, nodeId)}:${clientId}`;
   }
 
+  public static GET_PLAYER_CLIENT_LATENCY_TEST(sessionId: string) {
+    return `latency-test:${sessionId}`;
+  }
+
   constructor(
     private readonly logger: Logger,
     private readonly config: ConfigService,
@@ -90,8 +86,6 @@ export class SocketsGateway {
           break;
       }
     });
-
-    void this.setupNode();
   }
 
   @SubscribeMessage("ping")
@@ -137,16 +131,19 @@ export class SocketsGateway {
 
         client.id = uuidv4();
         client.user = request.user;
+        client.sessionId = request.session.id;
         client.node = this.nodeId;
 
         this.clients.set(client.id, client);
 
         await this.updateClient(client.user.steam_id, client.id);
 
+        await this.matchmaking.cancelOffline(client.user.steam_id);
+
         await this.sendPeopleOnline();
         await this.matchmaking.sendRegionStats(client.user);
         await this.matchmakingLobbyService.sendQueueDetailsToPlayer(
-          client.user,
+          client.user.steam_id,
         );
 
         client.on("close", async () => {
@@ -169,24 +166,7 @@ export class SocketsGateway {
 
             await this.sendPeopleOnline();
 
-            // GIVE THEM A DELAY
-            // this.matchMaking.leaveQueue(client);
-
-            // await this.hasura.mutation({
-            //   delete_lobby_players: {
-            //     __args: {
-            //       where: {
-            //         steam_id: {
-            //           _eq: client.user.steam_id,
-            //         },
-            //         status: {
-            //           _eq: "Accepted",
-            //         },
-            //       },
-            //     },
-            //     __typename: true,
-            //   },
-            // });
+            this.matchmaking.markOffline(client.user.steam_id);
           }
         });
       });
@@ -285,24 +265,4 @@ export class SocketsGateway {
 
     this.clients.delete(clientId);
   }
-
-  private async setupNode() {
-    await this.redis.sadd(
-      SocketsGateway.GET_AVAILABLE_NODES_KEY(),
-      this.nodeId,
-    );
-    const markOnline = async () => {
-      await this.redis.set(
-        SocketsGateway.GET_NODE_STATUS_KEY(this.nodeId),
-        "true",
-        "EX",
-        60,
-      );
-    };
-
-    // await markOnline();
-    // setInterval(async () => {
-    //   await markOnline();
-    // }, 30 * 1000);
-  }
 }