feature: playcast relay in redis

package.json

@@ -93,6 +93,7 @@
     "@nestjs/schematics": "^11.0.5",
     "@nestjs/testing": "^11.1.3",
     "@testcontainers/postgresql": "^12.0.4",
+    "testcontainers": "^12.0.4",
     "@types/archiver": "^6.0.2",
     "@types/express": "^5.0.0",
     "@types/express-session": "^1.18.0",

src/configs/redis.ts

@@ -13,6 +13,20 @@ export default (): {
           : undefined,
         password: process.env.REDIS_PASSWORD,
       },
+      // Playcast relay traffic: its own socket so fragment payloads never queue
+      // behind session lookups, and failing fast so an outage answers viewers
+      // and game servers with a 503 instead of holding their requests open.
+      relay: {
+        db: 1,
+        host: process.env.REDIS_HOST || "redis",
+        port: process.env.REDIS_SERVICE_PORT
+          ? parseInt(process.env.REDIS_SERVICE_PORT)
+          : undefined,
+        password: process.env.REDIS_PASSWORD,
+        enableOfflineQueue: false,
+        maxRetriesPerRequest: 1,
+        commandTimeout: 2000,
+      },
       sub: {
         db: 1,
         host: process.env.REDIS_HOST || "redis",

src/configs/types/RedisConfig.ts

@@ -6,6 +6,9 @@ export type RedisConfig = {
       host: string;
       port: number;
       password: string;
+      enableOfflineQueue?: boolean;
+      maxRetriesPerRequest?: number | null;
+      commandTimeout?: number;
     }
   >;
 };

src/matches/jobs/StopMatchBroadcast.ts

@@ -30,7 +30,7 @@ export class StopMatchBroadcast extends WorkerHost {
       return;
     }
 
-    this.matchRelay.removeBroadcast(matchId);
+    await this.matchRelay.removeBroadcast(matchId);
 
     if (!(await this.gameStreamer.stopLiveIfRunning(matchId))) {
       return;

src/matches/match-relay/match-relay.controller.ts

@@ -1,92 +1,84 @@
-import { Controller, Get, Post, Req, Res, Param } from "@nestjs/common";
+import { Controller, Get, Post, Req, Res, Param, Logger } from "@nestjs/common";
 import { Request, Response } from "express";
 import { MatchRelayService } from "./match-relay.service";
+import { FragmentField } from "./types/fragment.types";
 
 @Controller("match-relay/:id")
 export class MatchRelayController {
-  constructor(private readonly matchRelayService: MatchRelayService) {}
+  constructor(
+    private readonly logger: Logger,
+    private readonly matchRelayService: MatchRelayService,
+  ) {}
 
   @Get("sync")
-  public handleSyncGet(
+  public async handleSyncGet(
     @Param("id") matchId: string,
     @Req() request: Request,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getSyncInfo(request, response, matchId);
+    await this.relay(matchId, response, () =>
+      this.matchRelayService.getSyncInfo(request, response, matchId),
+    );
   }
 
   @Get(":fragment/start")
-  public handleGetStart(
+  public async handleGetStart(
     @Param("id") matchId: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getStart(response, matchId, parseInt(fragment));
+    await this.relay(matchId, response, () =>
+      this.matchRelayService.getStart(response, matchId, parseInt(fragment)),
+    );
   }
 
   @Get(":fragment/full")
-  public handleGetFull(
+  public async handleGetFull(
     @Param("id") matchId: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getFragment(
-      response,
-      matchId,
-      parseInt(fragment),
-      "full",
-    );
+    await this.getFragment(response, matchId, fragment, "full");
   }
 
   @Get(":fragment/delta")
-  public handleGetDelta(
+  public async handleGetDelta(
     @Param("id") matchId: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getFragment(
-      response,
-      matchId,
-      parseInt(fragment),
-      "delta",
-    );
+    await this.getFragment(response, matchId, fragment, "delta");
   }
 
   @Get(":token/:fragment/start")
-  public handleGetStartWithToken(
+  public async handleGetStartWithToken(
     @Param("id") matchId: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getStart(response, matchId, parseInt(fragment));
+    await this.relay(matchId, response, () =>
+      this.matchRelayService.getStart(response, matchId, parseInt(fragment)),
+    );
   }
 
   @Get(":token/:fragment/full")
-  public handleGetFullWithToken(
+  public async handleGetFullWithToken(
     @Param("id") matchId: string,
+    @Param("token") token: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getFragment(
-      response,
-      matchId,
-      parseInt(fragment),
-      "full",
-    );
+    await this.getFragment(response, matchId, fragment, "full", token);
   }
 
   @Get(":token/:fragment/delta")
-  public handleGetDeltaWithToken(
+  public async handleGetDeltaWithToken(
     @Param("id") matchId: string,
+    @Param("token") token: string,
     @Param("fragment") fragment: string,
     @Res() response: Response,
   ) {
-    this.matchRelayService.getFragment(
-      response,
-      matchId,
-      parseInt(fragment),
-      "delta",
-    );
+    await this.getFragment(response, matchId, fragment, "delta", token);
   }
 
   @Post(":token/:fragment/start")
@@ -97,14 +89,7 @@ export class MatchRelayController {
     @Req() request: Request,
     @Res() response: Response,
   ) {
-    this.matchRelayService.postField(
-      request,
-      response,
-      token,
-      "start",
-      matchId,
-      parseInt(fragment),
-    );
+    await this.postField(request, response, token, "start", matchId, fragment);
   }
 
   @Post(":token/:fragment/full")
@@ -115,14 +100,7 @@ export class MatchRelayController {
     @Req() request: Request,
     @Res() response: Response,
   ) {
-    this.matchRelayService.postField(
-      request,
-      response,
-      token,
-      "full",
-      matchId,
-      parseInt(fragment),
-    );
+    await this.postField(request, response, token, "full", matchId, fragment);
   }
 
   @Post(":token/:fragment/delta")
@@ -133,13 +111,62 @@ export class MatchRelayController {
     @Req() request: Request,
     @Res() response: Response,
   ) {
-    this.matchRelayService.postField(
-      request,
-      response,
-      token,
-      "delta",
-      matchId,
-      parseInt(fragment),
+    await this.postField(request, response, token, "delta", matchId, fragment);
+  }
+
+  private getFragment(
+    response: Response,
+    matchId: string,
+    fragment: string,
+    field: FragmentField,
+    token?: string,
+  ) {
+    return this.relay(matchId, response, () =>
+      this.matchRelayService.getFragment(
+        response,
+        matchId,
+        parseInt(fragment),
+        field,
+        token,
+      ),
+    );
+  }
+
+  private postField(
+    request: Request,
+    response: Response,
+    token: string,
+    field: FragmentField,
+    matchId: string,
+    fragment: string,
+  ) {
+    return this.relay(matchId, response, () =>
+      this.matchRelayService.postField(
+        request,
+        response,
+        token,
+        field,
+        matchId,
+        parseInt(fragment),
+      ),
     );
   }
+
+  private async relay(
+    matchId: string,
+    response: Response,
+    handle: () => Promise<void>,
+  ) {
+    try {
+      await handle();
+    } catch (error) {
+      this.logger.error(
+        `[${matchId}] relay request failed: ${(error as Error)?.message}`,
+      );
+      if (!response.headersSent) {
+        response.writeHead(503, { "Cache-Control": "no-store" });
+      }
+      response.end();
+    }
+  }
 }