fix: api event-dedup ordering, news stream errors, redis parse guard (#532, #546, part of #557)

src/matches/match-events.gateway.spec.ts

@@ -0,0 +1,76 @@
+// Isolate the gateway from its heavy DI imports; we only exercise the
+// dedup/processing ordering logic in handleMatchEvent.
+jest.mock("./events", () => ({ MatchEvents: { testEvent: class {} } }));
+jest.mock("src/hasura/hasura.service", () => ({ HasuraService: class {} }));
+jest.mock("src/cache/cache.service", () => ({ CacheService: class {} }));
+
+import { MatchEventsGateway } from "./match-events.gateway";
+
+function makeGateway(opts: { cacheHit?: boolean; processImpl?: () => any }) {
+  const cache = {
+    has: jest.fn().mockResolvedValue(opts.cacheHit ?? false),
+    put: jest.fn().mockResolvedValue(undefined),
+  };
+  const processor = {
+    setData: jest.fn(),
+    process: jest.fn().mockImplementation(opts.processImpl ?? (async () => {})),
+  };
+  const moduleRef = { resolve: jest.fn().mockResolvedValue(processor) };
+  const logger = {
+    warn: jest.fn(),
+    error: jest.fn(),
+    log: jest.fn(),
+    debug: jest.fn(),
+  };
+  const gateway = new MatchEventsGateway(
+    logger as any,
+    moduleRef as any,
+    {} as any,
+    cache as any,
+  );
+  return { gateway, cache, processor, logger };
+}
+
+const message = {
+  matchId: "match-1",
+  messageId: "msg-1",
+  data: { event: "testEvent", data: {} },
+};
+
+describe("MatchEventsGateway.handleMatchEvent dedup", () => {
+  it("marks the event processed only after process() succeeds", async () => {
+    const { gateway, cache, processor } = makeGateway({});
+    const result = await gateway.handleMatchEvent(message as any);
+
+    expect(processor.process).toHaveBeenCalledTimes(1);
+    expect(cache.put).toHaveBeenCalledTimes(1);
+    // Ordering: process resolved before the dedup entry was written.
+    expect(processor.process.mock.invocationCallOrder[0]).toBeLessThan(
+      cache.put.mock.invocationCallOrder[0],
+    );
+    expect(result).toBe("msg-1");
+  });
+
+  it("does NOT write the dedup entry when process() throws (so redelivery retries)", async () => {
+    const { gateway, cache, processor } = makeGateway({
+      processImpl: async () => {
+        throw new Error("hasura down");
+      },
+    });
+
+    await expect(gateway.handleMatchEvent(message as any)).rejects.toThrow(
+      "hasura down",
+    );
+    expect(processor.process).toHaveBeenCalledTimes(1);
+    expect(cache.put).not.toHaveBeenCalled();
+  });
+
+  it("short-circuits on a dedup hit without processing", async () => {
+    const { gateway, cache, processor } = makeGateway({ cacheHit: true });
+    const result = await gateway.handleMatchEvent(message as any);
+
+    expect(result).toBe("msg-1");
+    expect(processor.process).not.toHaveBeenCalled();
+    expect(cache.put).not.toHaveBeenCalled();
+  });
+});

src/matches/match-events.gateway.ts

@@ -115,8 +115,6 @@ export class MatchEventsGateway {
       return messageId;
     }
 
-    await this.cache.put(cacheKey, true, 10);
-
     const { data, event } = message.data;
 
     this.logger.debug(
@@ -127,7 +125,7 @@ export class MatchEventsGateway {
 
     if (!Processor) {
       this.logger.warn("unable to find event handler", event);
-      return;
+      return messageId;
     }
 
     const processor =
@@ -138,16 +136,21 @@ export class MatchEventsGateway {
     try {
       await processor.process();
     } catch (error) {
+      // Do NOT write the dedup entry on failure: leave the key absent so the
+      // game server's redelivery of this messageId is reprocessed instead of
+      // being silently swallowed by the dedup short-circuit for its TTL.
       this.logger.error(
-        `[${matchId}] error processing game event ${event} (messageId=${messageId}): ${error.message}`,
-        error.stack,
+        `[${matchId}] error processing game event ${event} (messageId=${messageId}): ${
+          (error as Error)?.message
+        }`,
+        (error as Error)?.stack,
       );
-      // withhold the ack and clear the dedup key so the game server's retry
-      // re-processes instead of being swallowed by the cache
-      await this.cache.forget(cacheKey);
-      return;
+      throw error;
     }
 
+    // Mark processed only after success.
+    await this.cache.put(cacheKey, true, 10);
+
     return messageId;
   }
 }

src/news/news.controller.ts

@@ -12,6 +12,7 @@ import {
   FileTypeValidator,
   ForbiddenException,
   NotFoundException,
+  Logger,
 } from "@nestjs/common";
 import { FileInterceptor } from "@nestjs/platform-express";
 import { Request, Response } from "express";
@@ -25,6 +26,8 @@ import { NewsService } from "./news.service";
 
 @Controller("news")
 export class NewsController {
+  private readonly logger = new Logger(NewsController.name);
+
   constructor(
     private readonly news: NewsService,
     private readonly system: SystemService,
@@ -126,6 +129,25 @@ export class NewsController {
       res.setHeader("ETag", result.etag);
     }
 
+    // Without these handlers an S3 stream error would throw an unhandled
+    // 'error' event (crashing the process), and a client that disconnects
+    // mid-download would leave the upstream stream open (fd leak).
+    result.stream.on("error", (error: Error) => {
+      this.logger.error(
+        `error streaming news image ${filename}: ${error?.message}`,
+        error?.stack,
+      );
+      result.stream.destroy();
+      if (!res.headersSent) {
+        res.status(500).end();
+      } else {
+        res.destroy();
+      }
+    });
+    res.on("close", () => {
+      result.stream.destroy();
+    });
+
     result.stream.pipe(res);
   }
 

src/sockets/sockets.service.ts

@@ -41,7 +41,32 @@ export class SocketsService {
     void sub.subscribe("broadcast-message");
     void sub.subscribe("send-message-to-steam-id");
     sub.on("message", (channel, message) => {
-      const { steamId, event, data } = JSON.parse(message) as {
+      let parsed: unknown;
+      try {
+        parsed = JSON.parse(message);
+      } catch (error) {
+        // A malformed payload on the pub/sub channel must not take the pod
+        // down: this handler runs outside any request lifecycle, so a throw
+        // here is an uncaught exception.
+        this.logger.error(
+          `failed to parse pub/sub message on ${channel}: ${
+            (error as Error)?.message
+          }`,
+        );
+        return;
+      }
+
+      // JSON.parse succeeds for non-objects too (e.g. the literal `null`, which
+      // typeof-reports as "object"); guard before destructuring so a valid but
+      // non-object payload can't throw a TypeError here.
+      if (parsed === null || typeof parsed !== "object") {
+        this.logger.error(
+          `ignoring non-object pub/sub message on ${channel}`,
+        );
+        return;
+      }
+
+      const { steamId, event, data } = parsed as {
         steamId: string;
         event: string;
         data: unknown;