feat: add lobby sessions, join, leave, and broadcast module

nohub/.env.example

@@ -63,7 +63,9 @@ NOHUB_SESSIONS_MAX_COUNT=262144
 # Maximum number of sessions from the same IP address ( disregarding port )
 # Set to 0 to disable this limit
 NOHUB_SESSIONS_MAX_PER_ADDRESS=64
+# Set to true to create session ids as number
+NOHUB_SESSIONS_ID_USE_NUMBER=false
 
 # Other =======================================================================
 # Logging level - silent, trace, debug, info, warn, error, fatal
-NOHUB_LOG_LEVEL=info
+NOHUB_LOG_LEVEL=info

nohub/spec/api/sessions.api.spec.ts

@@ -23,4 +23,20 @@ describe("Sessions API", () => {
       expect(reply.text).not.toBeEmpty();
     });
   });
+
+  describe("getid", () => {
+    test("should respond", async () => {
+      const reply = await api
+        .client()
+        .send({
+          name: "getid",
+          isRequest: true,
+          requestId: "",
+        })
+        .onReply();
+
+      expect(reply.isSuccessResponse).toBeTrue();
+      expect(reply.text).not.toBeEmpty();
+    });
+  });
 });

nohub/spec/broadcast/broadcast.service.spec.ts

@@ -0,0 +1,125 @@
+import { beforeEach, describe, expect, mock, test } from "bun:test";
+import { Games, Lobbies, Sessions } from "@spec/fixtures";
+import { mockSocket } from "@spec/sessions/session.api.spec";
+import { BroadcastService } from "@src/broadcast/broadcast.service";
+import { readDefaultConfig } from "@src/config";
+import { DataNotFoundError } from "@src/errors";
+import { NohubEventBus } from "@src/events";
+import { GameRepository } from "@src/games/game.repository";
+import { LobbyApi } from "@src/lobbies/lobby.api";
+import { LobbyEventBus } from "@src/lobbies/lobby.events";
+import { LobbyRepository } from "@src/lobbies/lobby.repository";
+import { LobbyService } from "@src/lobbies/lobby.service";
+import type { NohubReactor } from "@src/nohub";
+import type { SessionData } from "@src/sessions/session";
+import { SessionApi } from "@src/sessions/session.api";
+import { SessionRepository } from "@src/sessions/session.repository";
+import type { Socket } from "bun";
+
+let reactor: NohubReactor;
+
+let sessionRepository: SessionRepository;
+let sessionApi: SessionApi;
+let lobbyRepository: LobbyRepository;
+let lobbyService: LobbyService;
+let lobbyApi: LobbyApi;
+
+let broadcastService: BroadcastService;
+
+let daveSocket: Socket<SessionData>;
+let ericSocket: Socket<SessionData>;
+
+describe("BroadcastService", () => {
+  beforeEach(() => {
+    reactor = {
+      send: mock(() => ({})),
+    } as unknown as NohubReactor;
+
+    sessionRepository = new SessionRepository();
+    const gameLookup = new GameRepository();
+    Games.insert(gameLookup);
+    Sessions.insert(sessionRepository);
+
+    sessionApi = new SessionApi(
+      sessionRepository,
+      new LobbyRepository(),
+      gameLookup,
+      new NohubEventBus(),
+      readDefaultConfig().sessions,
+    );
+
+    lobbyRepository = new LobbyRepository();
+    lobbyService = new LobbyService(
+      lobbyRepository,
+      readDefaultConfig().lobbies,
+      new LobbyEventBus(),
+    );
+
+    lobbyApi = new LobbyApi(lobbyRepository, lobbyService, () => undefined);
+
+    Lobbies.insert(lobbyRepository);
+
+    // create sessions using API
+    daveSocket = mockSocket(Sessions.dave.address);
+    ericSocket = mockSocket(Sessions.eric.address);
+
+    sessionApi.openSession(daveSocket);
+    sessionApi.openSession(ericSocket);
+
+    broadcastService = new BroadcastService(() => reactor, sessionRepository);
+  });
+
+  describe("unicast", () => {
+    test("should send command to session", () => {
+      const sessionId = daveSocket.data.id;
+      broadcastService.unicast(sessionId, { name: "command" });
+
+      expect(reactor.send).toHaveBeenCalled();
+      // Verify it was called with the correct socket
+      expect(reactor.send).toHaveBeenCalledWith(daveSocket, {
+        name: "command",
+      });
+    });
+
+    test("should throw if session not found", () => {
+      expect(() =>
+        broadcastService.unicast("unknown", { name: "command" }),
+      ).toThrow(DataNotFoundError);
+    });
+  });
+
+  describe("broadcast", () => {
+    test("should send to all participants", () => {
+      const lobby = lobbyApi.create(Sessions.dave.address, daveSocket.data);
+      lobbyApi.join(lobby.id, ericSocket.data);
+
+      broadcastService.broadcast(lobby, { name: "command" });
+
+      // Should broadcast to both
+      expect(reactor.send).toHaveBeenCalledTimes(2);
+      expect(reactor.send).toHaveBeenCalledWith(daveSocket, {
+        name: "command",
+      });
+      expect(reactor.send).toHaveBeenCalledWith(ericSocket, {
+        name: "command",
+      });
+    });
+
+    test("should skip missing sessions", () => {
+      const lobby = lobbyApi.create(Sessions.dave.address, daveSocket.data);
+
+      // Join
+      lobbyApi.join(lobby.id, ericSocket.data);
+      // Close (leave)
+      sessionApi.closeSession(ericSocket);
+
+      broadcastService.broadcast(lobby, { name: "command" });
+
+      // Should broadcast to just 1 particpant
+      expect(reactor.send).toHaveBeenCalledTimes(1);
+      expect(reactor.send).toHaveBeenCalledWith(daveSocket, {
+        name: "command",
+      });
+    });
+  });
+});

nohub/spec/fixtures.ts

@@ -105,6 +105,7 @@ export const Lobbies = {
       ["player-count", "8"],
       ["player-capacity", "12"],
     ]),
+    participants: [],
   }),
 
   coolLobby: lobbyFixture({
@@ -119,6 +120,7 @@ export const Lobbies = {
       ["player-count", "9"],
       ["player-capacity", "16"],
     ]),
+    participants: [],
   }),
 
   mithrilParty: lobbyFixture({
@@ -133,6 +135,21 @@ export const Lobbies = {
       ["player-count", "4"],
       ["player-capacity", "6"],
     ]),
+    participants: [],
+  }),
+
+  pamParticipantsLobby: lobbyFixture({
+    id: "GzF2zDhX",
+    owner: Sessions.pam.id,
+    address: Addresses.pam,
+    isVisible: true,
+    isLocked: false,
+    data: new Map([
+      ["name", "Pam's Participants Lobby"],
+      ["player-count", "8"],
+      ["player-capacity", "12"],
+    ]),
+    participants: [Sessions.pam.id, Sessions.ingrid.id],
   }),
 
   all(): Lobby[] {

nohub/spec/lobbies/lobby.repository.spec.ts

@@ -30,4 +30,28 @@ describe("LobbyRepository", () => {
       );
     });
   });
+  describe("removeLobbiesOf", () => {
+    test("should remove lobbies owned by session", () => {
+      const results = [...lobbyRepository.removeLobbiesOf(Sessions.dave.id)];
+      expect(results).toHaveLength(1);
+      expect(results[0]).toEqual(Lobbies.davesLobby);
+      expect(lobbyRepository.has(Lobbies.davesLobby.id)).toBeFalse();
+    });
+
+    test("should remove session from participants if not owner", () => {
+      expect([...lobbyRepository.listLobbiesFor(Sessions.pam)]).toEqual([
+        Lobbies.pamParticipantsLobby,
+      ]);
+      // Remove a participant who does not own a lobby.
+      // It does not remove the lobby, just the participant.
+      const results = [...lobbyRepository.removeLobbiesOf(Sessions.ingrid.id)];
+      expect(results).toHaveLength(0);
+      expect(
+        lobbyRepository.require(Lobbies.pamParticipantsLobby.id).participants,
+      ).toContain(Sessions.pam.id);
+      expect(
+        lobbyRepository.require(Lobbies.pamParticipantsLobby.id).participants,
+      ).not.toContain(Sessions.ingrid.id);
+    });
+  });
 });

nohub/spec/lobbies/lobby.service.spec.ts

@@ -46,6 +46,7 @@ describe("LobbyService", () => {
         isVisible: true,
         isLocked: false,
         data: lobbyData,
+        participants: [Sessions.dave.id],
       };
 
       const lobby = lobbyService.create(
@@ -212,6 +213,34 @@ describe("LobbyService", () => {
     });
   });
 
+  describe("leave", () => {
+    test("should leave lobby", () => {
+      const lobby = lobbyService.create(
+        Addresses.dave,
+        new Map(),
+        Sessions.dave,
+      );
+
+      lobbyService.join(lobby, Sessions.eric);
+      expect(lobby.participants).toContain(Sessions.eric.id);
+
+      lobbyService.leave(lobby, Sessions.eric);
+      expect(lobby.participants).not.toContain(Sessions.eric.id);
+    });
+
+    test("should throw if not in lobby", () => {
+      expect(() =>
+        lobbyService.leave(Lobbies.davesLobby, Sessions.eric),
+      ).toThrow(InvalidCommandError);
+    });
+
+    test("should throw if owner tries to leave", () => {
+      expect(() =>
+        lobbyService.leave(Lobbies.davesLobby, Sessions.dave),
+      ).toThrow(InvalidCommandError);
+    });
+  });
+
   describe("setData", () => {
     test("should replace lobby data", () => {
       const newData = Lobbies.coolLobby.data;

nohub/spec/sessions/session.api.spec.ts

@@ -104,6 +104,18 @@ describe("SessionApi", () => {
           sessionApi.openSession(mockSocket(Sessions.dave.address)),
         ).not.toThrow();
     });
+
+    test("should create session with number", () => {
+      // Don't need fixtures
+      sessionRepository.clear();
+
+      // Set use number
+      config.idUseNumber = true;
+
+      // Open 1 session and expect only numbers
+      sessionApi.openSession(mockSocket(Sessions.dave.address));
+      expect([...sessionRepository.list()][0].id).toMatch(/^[0-9]+$/);
+    });
   });
 
   describe("setGame", () => {
@@ -120,7 +132,7 @@ describe("SessionApi", () => {
   });
 });
 
-function mockSocket(address: string): Socket<SessionData> {
+export function mockSocket(address: string): Socket<SessionData> {
   return {
     remoteAddress: address,
     write: (

nohub/src/broadcast/broadcast.module.ts

@@ -0,0 +1,25 @@
+import type { Module } from "@src/module";
+import type { NohubReactor } from "@src/nohub";
+import type { SessionModule } from "@src/sessions/session.module";
+import { BroadcastService } from "./broadcast.service";
+
+export class BroadcastModule implements Module {
+  readonly broadcastService: BroadcastService;
+  private reactor?: NohubReactor;
+
+  constructor(sessionModule: SessionModule) {
+    this.broadcastService = new BroadcastService(
+      () => this.provideReactor(),
+      sessionModule.sessionRepository,
+    );
+  }
+
+  configure(reactor: NohubReactor) {
+    this.reactor = reactor;
+  }
+
+  private provideReactor(): NohubReactor {
+    if (!this.reactor) throw new Error("Missing Reactor instance!");
+    return this.reactor;
+  }
+}

nohub/src/broadcast/broadcast.service.ts

@@ -0,0 +1,37 @@
+import type { CommandSpec, Exchange } from "@foxssake/trimsock-js";
+import { DataNotFoundError } from "@src/errors";
+import type { Lobby } from "@src/lobbies/lobby";
+import type { NohubReactor } from "@src/nohub";
+import type { SessionId, SessionSocket } from "@src/sessions/session";
+import type { SessionRepository } from "@src/sessions/session.repository";
+
+export class BroadcastService {
+  constructor(
+    private reactor: () => NohubReactor,
+    private sessionRepository: SessionRepository,
+  ) {}
+
+  unicast(sessionId: string, command: CommandSpec): Exchange<SessionSocket> {
+    const session = this.sessionRepository.find(sessionId);
+    if (!session?.socket)
+      throw new DataNotFoundError(`No connection to session#${session?.id}!`); // TODO: Probably a more specific exception
+
+    return this.reactor().send(session.socket, command);
+  }
+
+  broadcast(
+    lobby: Lobby,
+    command: CommandSpec,
+  ): Map<SessionId, Exchange<SessionSocket>> {
+    const result = new Map();
+
+    for (const sessionId of lobby.participants) {
+      const session = this.sessionRepository.find(sessionId);
+      if (!session) continue; // Shouldn't happen, unless lobby participants are not cleared up on client disconnect
+
+      result.set(sessionId, this.unicast(sessionId, command));
+    }
+
+    return result;
+  }
+}

nohub/src/config.ts

@@ -45,6 +45,7 @@ export function readConfig(env: ConfigEnv) {
       defaultGameId: env.NOHUB_LOBBIES_DEFAULT_GAME_ID,
       maxCount: integer(env.NOHUB_SESSIONS_MAX_COUNT) ?? 262144,
       maxPerAddress: integer(env.NOHUB_SESSIONS_MAX_PER_ADDRESS) ?? 64,
+      idUseNumber: bool(env.NOHUB_SESSIONS_ID_USE_NUMBER) ?? false,
     },
   };
 }

nohub/src/lobbies/lobby.api.ts

@@ -88,6 +88,15 @@ export class LobbyApi {
     return address;
   }
 
+  leave(id: string, session: SessionData): void {
+    this.logger.info({ session, lobbyId: id }, "Leaving lobby");
+
+    const lobby = this.lobbyRepository.requireInGame(id, session.gameId);
+    this.lobbyService.leave(lobby, session);
+
+    this.logger.info({ session, lobby }, "Successfully left lobby");
+  }
+
   setData(id: string, data: Map<string, string>, session: SessionData): void {
     this.logger.info({ lobbyId: id, session, data }, "Updating session data");
 

nohub/src/lobbies/lobby.module.ts

@@ -92,6 +92,14 @@ export class LobbyModule implements Module {
         const address = this.lobbyApi.join(lobbyId, session);
         xchg.reply({ params: [address] });
       })
+      .on("lobby/leave", (cmd, xchg) => {
+        requireRequest(cmd);
+        const lobbyId = cmd.requireText();
+        const session = sessionOf(xchg);
+
+        this.lobbyApi.leave(lobbyId, session);
+        xchg.reply({ text: "ok" });
+      })
       .on("lobby/set-data", (cmd, xchg) => {
         requireRequest(cmd);
         const lobbyId = requireSingleParam(cmd, "Missing lobby ID!");