feat(backend): save every game and keep live games across restarts

- Games record start and move times, spectator peak and distinct count
- Games end as completed, left (the leaver loses), timeout (5 minutes
  offline on your turn while the opponent is online) or inactive (both
  offline for 30 minutes); everyone gets on_game_end
- Connection info carries offlineSince for the countdown
- A new game is rejected while the user has one in progress
- Live games are written to Valkey after every change and restored on
  startup; finished games are saved to Postgres with public ids, and
  saves that fail are kept and retried every 30 seconds
- Clients get a trimmed game without spectator ids
- Startup waits for Valkey, restores games, then accepts connections;
  migrations are retried while Postgres is unreachable
This commit is contained in:
Halit Aksoy 2026-10-16 00:18:33 +03:00
parent f6136be007
commit 7239bbb6ed
13 changed files with 865 additions and 74 deletions

View File

@ -1,4 +1,4 @@
import { CommonMancalaGame, MancalaGame } from "mancala.js";
import { CommonMancalaGame } from "mancala.js";
import { RTMT } from "../rtmt/rtmt";
import { GameStore } from "./gamestore/GameStore";
import { generateKey } from "../util/key_factory";
@ -9,27 +9,59 @@ import {
channel_listen_game_events,
channel_new_game,
channel_on_game_crashed,
channel_on_game_end,
channel_on_game_start,
channel_on_game_update,
channel_on_game_user_leave,
channel_on_new_game_rejected,
channel_on_user_connection_change,
channel_unlisten_game_events
channel_unlisten_game_events,
Game,
GameEndMessage,
GameMove,
GameUsersConnectionInfo,
INACTIVE_TIMEOUT_MS,
TURN_TIMEOUT_MS,
UserConnectionInfo,
} from "@mancala/core";
import { GameMove } from "@mancala/core";
import { GameCrashManager } from "./GameCrashManager";
import { MatchMaker } from "../matchmaker/MatchMaker";
import { GameUsersConnectionInfo, UserConnectionInfo } from "@mancala/core";
import { BackendGame, PlayerInfo } from "../models/Game";
import { LiveGameStore } from "../live/LiveGameStore";
import { GameRecorder } from "./GameRecorder";
import { fromSnapshot, GameEnding, getResult, isLiveGameSnapshot, toGameRecord, toSnapshot } from "./GameSerialization";
const TIMEOUT_CHECK_INTERVAL_MS = 10_000;
// What players and spectators receive; leaves out spectator ids and other server-side state.
export type ClientGame = Game & { startedAt: number, moveTimes: number[], spectatorCount: number };
export class GameManager {
gameStore: GameStore;
rtmt: RTMT;
matchMaker: MatchMaker;
private readonly liveGameStore: LiveGameStore;
private readonly recorder: GameRecorder;
private readonly clock: () => number;
// When each player in a live game went offline; cleared when they reconnect.
private readonly offlineSince = new Map<string, number>();
private readonly liveGamesListeners: (() => void)[] = [];
private timeoutTimer?: NodeJS.Timeout;
constructor(params: { gameStore: GameStore, rtmt: RTMT, matchMaker: MatchMaker }) {
constructor(params: {
gameStore: GameStore,
rtmt: RTMT,
matchMaker: MatchMaker,
liveGameStore: LiveGameStore,
recorder: GameRecorder,
clock?: () => number,
}) {
this.gameStore = params.gameStore;
this.rtmt = params.rtmt;
this.matchMaker = params.matchMaker;
this.liveGameStore = params.liveGameStore;
this.recorder = params.recorder;
this.clock = params.clock ?? Date.now;
this.initialize();
}
@ -39,8 +71,26 @@ export class GameManager {
this.listenUserConnectionChange();
}
// Called whenever a live game starts, changes or ends.
public onLiveGamesChanged(listener: () => void) {
this.liveGamesListeners.push(listener);
}
public startTimers() {
this.timeoutTimer = setInterval(() => this.checkTimeouts(), TIMEOUT_CHECK_INTERVAL_MS);
}
public stopTimers() {
if (this.timeoutTimer) clearInterval(this.timeoutTimer);
}
private listenRtmtMessages() {
this.rtmt.listenMessage(channel_new_game, (userKey: string, message: Object) => {
const activeGame = this.gameStore.getGameByUser(userKey);
if (activeGame) {
this.rtmt.sendMessage(userKey, channel_on_new_game_rejected, { gameId: activeGame.id });
return;
}
this.matchMaker.join(userKey);
});
@ -69,11 +119,16 @@ export class GameManager {
const game = this.gameStore.getGameByUser(userKey);
if (game) {
try {
const movesBefore = game.mancalaGame.history.length;
game.mancalaGame.moveByPlayerPit(userKey, gameMove.index);
if (game.mancalaGame.history.length > movesBefore) game.moveTimes.push(this.clock());
game.gameUsersConnectionInfo = this.getGameUsersConnectionInfo(game);
this.sendMessageToPlayersAndSpectators(game, channel_on_game_update, game);
this.sendMessageToPlayersAndSpectators(game, channel_on_game_update, this.toClientGame(game));
if (game.mancalaGame.state == "ended") {
this.onGameEnd(game);
this.endGame(game, { endReason: "completed", winnerId: game.mancalaGame.getWonPlayerId() });
} else {
this.saveLiveGame(game);
this.notifyLiveGamesChanged();
}
} catch (err: any) {
this.onGameError(game, err)
@ -93,11 +148,86 @@ export class GameManager {
private onPlayerLeave(userKey: string) {
const game = this.gameStore.getGameByUser(userKey);
if (game) {
this.deleteGame(game);
this.sendMessageToPlayersAndSpectators(game, channel_on_game_user_leave, userKey)
this.endGame(game, { endReason: "left", winnerId: this.getOpponentId(game, userKey), leftUserId: userKey });
}
}
// Ends a live game for any reason: tells everyone, forgets it and saves it.
private endGame(game: BackendGame, ending: GameEnding) {
if (this.gameStore.get(game.id) !== game) return;
this.deleteGame(game);
const endMessage: GameEndMessage = { gameId: game.id, result: getResult(game, ending), endReason: ending.endReason };
if (ending.winnerId !== undefined) endMessage.winnerId = ending.winnerId;
if (ending.leftUserId !== undefined) {
endMessage.leftUserId = ending.leftUserId;
this.sendMessageToPlayersAndSpectators(game, channel_on_game_user_leave, ending.leftUserId);
}
this.sendMessageToPlayersAndSpectators(game, channel_on_game_end, endMessage);
game.spectatorIds = [];
this.offlineSince.delete(game.mancalaGame.player1Id);
this.offlineSince.delete(game.mancalaGame.player2Id);
const record = toGameRecord(game, ending, this.clock());
this.liveGameStore.removeGame(game.id).catch((err) => this.logStoreError(err));
this.recorder.save(record);
this.notifyLiveGamesChanged();
}
// Applies the rules for players who stay away: the player whose turn it is loses after TURN_TIMEOUT_MS offline
// while the opponent is online; a game where both are offline for INACTIVE_TIMEOUT_MS ends without a winner.
public checkTimeouts() {
const now = this.clock();
for (const game of this.gameStore.getAll()) {
const { player1Id, player2Id, turnPlayerId } = game.mancalaGame;
const player1Offline = this.offlineSince.get(player1Id);
const player2Offline = this.offlineSince.get(player2Id);
if (player1Offline !== undefined && player2Offline !== undefined) {
if (now - Math.max(player1Offline, player2Offline) >= INACTIVE_TIMEOUT_MS) {
this.endGame(game, { endReason: "inactive" });
}
continue;
}
const turnPlayerOffline = this.offlineSince.get(turnPlayerId);
if (turnPlayerOffline !== undefined && now - turnPlayerOffline >= TURN_TIMEOUT_MS) {
this.endGame(game, { endReason: "timeout", winnerId: this.getOpponentId(game, turnPlayerId) });
}
}
}
// Puts the live games saved in Valkey back into memory, e.g. after a restart or deploy.
public async restoreLiveGames(): Promise<number> {
let snapshots: unknown[];
try {
snapshots = await this.liveGameStore.loadGames();
} catch (err) {
console.error("(GameManager) Could not load live games:", (err as Error).message);
return 0;
}
let restored = 0;
for (const snapshot of snapshots) {
if (!isLiveGameSnapshot(snapshot) || this.gameStore.get(snapshot.id)) continue;
// A game that ended but whose snapshot could not be removed must not come back.
if (await this.recorder.isRecorded(snapshot.id).catch(() => false)) {
this.liveGameStore.removeGame(snapshot.id).catch((err) => this.logStoreError(err));
continue;
}
const game = fromSnapshot(snapshot);
this.gameStore.set(game.id, game);
const now = this.clock();
for (const playerId of [game.mancalaGame.player1Id, game.mancalaGame.player2Id]) {
this.gameStore.setGameByUser(playerId, game);
if (!this.rtmt.isClientOnline(playerId)) this.offlineSince.set(playerId, now);
}
game.gameUsersConnectionInfo = this.getGameUsersConnectionInfo(game);
restored++;
}
if (restored > 0) {
console.info(`(GameManager) Restored ${restored} live game(s)`);
this.notifyLiveGamesChanged();
}
return restored;
}
private addSpectator(userId: string, gameId: string) {
const game = this.gameStore.get(gameId);
if (game) {
@ -105,6 +235,10 @@ export class GameManager {
const isAlreadySpectator = game.spectatorIds.find((value) => value === userId);
if (!isAlreadySpectator && !isUserPlayer) {
game.spectatorIds.push(userId);
if (!game.allSpectatorIds.includes(userId)) game.allSpectatorIds.push(userId);
game.peakSpectators = Math.max(game.peakSpectators, game.spectatorIds.length);
this.saveLiveGame(game);
this.notifyLiveGamesChanged();
}
}
}
@ -115,6 +249,7 @@ export class GameManager {
const userIdIndex = game.spectatorIds.findIndex((value) => value === userId);
if (userIdIndex >= 0) {
game.spectatorIds.splice(userIdIndex, 1);
this.notifyLiveGamesChanged();
}
}
}
@ -123,6 +258,11 @@ export class GameManager {
return game.mancalaGame.player1Id === userId || game.mancalaGame.player2Id === userId;
}
private getOpponentId(game: BackendGame, userId: string): string {
const { player1Id, player2Id } = game.mancalaGame;
return userId === player1Id ? player2Id : player1Id;
}
private sendMessageToPlayersAndSpectators(game: BackendGame, channel: string, message: Object) {
const sendMessage = (userId: string) => this.rtmt.sendMessage(userId, channel, message);
sendMessage(game.mancalaGame.player1Id);
@ -130,9 +270,15 @@ export class GameManager {
game.spectatorIds.forEach((spectatorId) => sendMessage(spectatorId));
}
private onGameEnd(game: BackendGame) {
this.deleteGame(game);
game.spectatorIds = [];
public toClientGame(game: BackendGame): ClientGame {
return {
id: game.id,
mancalaGame: game.mancalaGame,
gameUsersConnectionInfo: game.gameUsersConnectionInfo,
startedAt: game.startedAt,
moveTimes: game.moveTimes,
spectatorCount: game.spectatorIds.length,
};
}
private listenOnPlayersPaired() {
@ -156,7 +302,11 @@ export class GameManager {
mancalaGame,
gameUsersConnectionInfo: this.getGameUsersConnectionInfoFromPlayers(players[userKey1], players[userKey2]),
spectatorIds: [],
allSpectatorIds: [],
peakSpectators: 0,
players,
startedAt: this.clock(),
moveTimes: [],
};
this.gameStore.set(game.id, game);
@ -172,12 +322,17 @@ export class GameManager {
public getGameUsersConnectionInfo(game: BackendGame): GameUsersConnectionInfo {
const { player1Id, player2Id } = game.mancalaGame;
return this.getGameUsersConnectionInfoFromPlayers(game.players[player1Id], game.players[player2Id]);
const player = (id: string): PlayerInfo => game.players[id] ?? { userId: id, isGuest: true };
return this.getGameUsersConnectionInfoFromPlayers(player(player1Id), player(player2Id));
}
private getGameUsersConnectionInfoFromPlayers(player1: PlayerInfo, player2: PlayerInfo): GameUsersConnectionInfo {
const toConnectionInfo = (player: PlayerInfo): UserConnectionInfo =>
({ ...player, isOnline: this.rtmt.isClientOnline(player.userId) });
const toConnectionInfo = (player: PlayerInfo): UserConnectionInfo => {
const info: UserConnectionInfo = { ...player, isOnline: this.rtmt.isClientOnline(player.userId) };
const offlineSince = this.offlineSince.get(player.userId);
if (!info.isOnline && offlineSince !== undefined) info.offlineSince = offlineSince;
return info;
};
return {
user1ConnectionInfo: toConnectionInfo(player1),
user2ConnectionInfo: toConnectionInfo(player2)
@ -185,13 +340,15 @@ export class GameManager {
}
public startGame(game: BackendGame) {
this.sendMessageToPlayersAndSpectators(game, channel_on_game_start, game);
this.sendMessageToPlayersAndSpectators(game, channel_on_game_start, this.toClientGame(game));
this.sendUserConnectionInfo(game);
this.saveLiveGame(game);
this.notifyLiveGamesChanged();
}
public sendUserConnectionInfo(game: BackendGame) {
const gameUsersConnectionInfo = this.getGameUsersConnectionInfo(game);
this.sendMessageToPlayersAndSpectators(game, channel_on_user_connection_change, gameUsersConnectionInfo);
game.gameUsersConnectionInfo = this.getGameUsersConnectionInfo(game);
this.sendMessageToPlayersAndSpectators(game, channel_on_user_connection_change, game.gameUsersConnectionInfo);
}
public deleteGame(game: BackendGame) {
@ -202,15 +359,31 @@ export class GameManager {
}
}
private saveLiveGame(game: BackendGame) {
this.liveGameStore.saveGame(game.id, toSnapshot(game)).catch((err) => this.logStoreError(err));
}
private logStoreError(err: Error) {
console.error("(GameManager) Live game store:", err.message);
}
private notifyLiveGamesChanged() {
this.liveGamesListeners.forEach((listener) => listener());
}
private listenUserConnectionChange() {
this.rtmt.listenOnClientConnectionChange((clientId: string, isOnline: boolean) => {
const game = this.gameStore.getGameByUser(clientId);
if (game) this.sendUserConnectionInfo(game);
if (game) {
if (isOnline) this.offlineSince.delete(clientId);
else if (!this.offlineSince.has(clientId)) this.offlineSince.set(clientId, this.clock());
this.sendUserConnectionInfo(game);
}
if (!isOnline) {
const speactatingGameIds = this.gameStore.getSpeactatingGameIdsByUser(clientId);
speactatingGameIds.forEach((gameId) => this.removeSpectator(clientId, gameId));
} else {
game && this.rtmt.sendMessage(clientId, channel_on_game_update, game);
game && this.rtmt.sendMessage(clientId, channel_on_game_update, this.toClientGame(game));
}
});
}

View File

@ -0,0 +1,71 @@
import { GameRecord } from "@mancala/core";
import { GameHistoryRepository } from "../history/GameHistoryRepository";
import { LiveGameStore } from "../live/LiveGameStore";
const RETRY_INTERVAL_MS = 30_000;
// Saves finished games. When the database is unreachable the record is kept (in Valkey, or in memory if Valkey is
// down too) and saved later, so a game is never lost because a dependency was briefly unavailable.
export class GameRecorder {
private readonly pending = new Map<string, GameRecord>();
private readonly savedListeners: ((record: GameRecord) => void)[] = [];
private retryTimer?: NodeJS.Timeout;
private retrying = false;
constructor(private readonly repository: GameHistoryRepository, private readonly liveGameStore: LiveGameStore) { }
onSaved(listener: (record: GameRecord) => void) {
this.savedListeners.push(listener);
}
async save(record: GameRecord): Promise<void> {
try {
await this.repository.save(record);
this.notifySaved(record);
} catch (err) {
console.error(`(GameRecorder) Could not save game ${record.id}, will retry:`, (err as Error).message);
this.pending.set(record.id, record);
await this.liveGameStore.savePendingRecord(record).catch(() => undefined);
}
}
// True if the game is saved or waiting to be saved. Throws if the database cannot be asked.
async isRecorded(gameId: string): Promise<boolean> {
return this.pending.has(gameId) || (await this.repository.get(gameId)) !== undefined;
}
async retryPending(): Promise<void> {
if (this.retrying) return;
this.retrying = true;
try {
const stored = await this.liveGameStore.loadPendingRecords().catch(() => []);
stored.forEach((record) => this.pending.set(record.id, record));
for (const record of [...this.pending.values()]) {
try {
await this.repository.save(record);
} catch (err) {
return; // still unreachable; try again later
}
this.pending.delete(record.id);
await this.liveGameStore.removePendingRecord(record.id).catch(() => undefined);
console.info(`(GameRecorder) Saved game ${record.id} after retrying`);
this.notifySaved(record);
}
} finally {
this.retrying = false;
}
}
start() {
this.retryPending();
this.retryTimer = setInterval(() => this.retryPending(), RETRY_INTERVAL_MS);
}
stop() {
if (this.retryTimer) clearInterval(this.retryTimer);
}
private notifySaved(record: GameRecord) {
this.savedListeners.forEach((listener) => listener(record));
}
}

View File

@ -0,0 +1,121 @@
import { EndReason, GameRecord, GameResult, MoveRecord } from "@mancala/core";
import { CommonMancalaGame, GameState, GameStep } from "mancala.js";
import { BackendGame, PlayerInfo } from "../models/Game";
import { publicUserId } from "./publicIds";
type HistoryEntry = { playerId: string, moveIndex: number, boardSnapshot: number[], gameSteps: GameStep[] };
// What is kept in Valkey for a live game; enough to rebuild it after a restart.
export interface LiveGameSnapshot {
version: 1;
id: string;
player1Id: string;
player2Id: string;
turnPlayerId: string;
state: GameState;
board: number[];
history: HistoryEntry[];
players: Record<string, PlayerInfo>;
startedAt: number;
moveTimes: number[];
peakSpectators: number;
allSpectatorIds: string[];
}
export interface GameEnding {
endReason: EndReason;
// Undefined when the game ends without a winner.
winnerId?: string;
leftUserId?: string;
}
export function toSnapshot(game: BackendGame): LiveGameSnapshot {
const { mancalaGame } = game;
return {
version: 1,
id: game.id,
player1Id: mancalaGame.player1Id,
player2Id: mancalaGame.player2Id,
turnPlayerId: mancalaGame.turnPlayerId,
state: mancalaGame.state,
board: mancalaGame.board.getStoneArray(),
history: mancalaGame.history as unknown as HistoryEntry[],
players: game.players,
startedAt: game.startedAt,
moveTimes: game.moveTimes,
peakSpectators: game.peakSpectators,
allSpectatorIds: game.allSpectatorIds,
};
}
export function isLiveGameSnapshot(value: unknown): value is LiveGameSnapshot {
const snapshot = value as LiveGameSnapshot;
return typeof snapshot === "object" && snapshot !== null && snapshot.version === 1
&& typeof snapshot.id === "string" && Array.isArray(snapshot.board) && snapshot.board.length === 14
&& Array.isArray(snapshot.history) && typeof snapshot.players === "object";
}
// Rebuilds a live game with real engine objects (rules, pits), so it can be played on.
export function fromSnapshot(snapshot: LiveGameSnapshot): BackendGame {
const mancalaGame = new CommonMancalaGame(snapshot.id, snapshot.player1Id, snapshot.player2Id);
snapshot.board.forEach((stoneCount, index) => (mancalaGame.board.pits[index].stoneCount = stoneCount));
mancalaGame.turnPlayerId = snapshot.turnPlayerId;
mancalaGame.state = snapshot.state;
mancalaGame.history.push(...snapshot.history);
const player = (id: string): PlayerInfo => snapshot.players[id] ?? { userId: id, isGuest: true };
return {
id: snapshot.id,
mancalaGame,
gameUsersConnectionInfo: {
user1ConnectionInfo: { ...player(snapshot.player1Id), isOnline: false },
user2ConnectionInfo: { ...player(snapshot.player2Id), isOnline: false },
},
spectatorIds: [],
allSpectatorIds: snapshot.allSpectatorIds,
peakSpectators: snapshot.peakSpectators,
players: snapshot.players,
startedAt: snapshot.startedAt,
moveTimes: snapshot.moveTimes,
};
}
export function getResult(game: BackendGame, ending: GameEnding): GameResult | null {
const { player1Id, player2Id } = game.mancalaGame;
if (ending.winnerId === player1Id) return "player1";
if (ending.winnerId === player2Id) return "player2";
return ending.endReason === "completed" ? "draw" : null;
}
// The record saved for a finished game; every id in it is a public id.
export function toGameRecord(game: BackendGame, ending: GameEnding, endedAt: number): GameRecord {
const { mancalaGame } = game;
const history = mancalaGame.history as unknown as HistoryEntry[];
const playerSummary = (id: string) => {
const player = game.players[id];
const summary = { userId: publicUserId(id), isGuest: player?.isGuest ?? true };
return player?.name === undefined ? summary : { ...summary, name: player.name };
};
const moves: MoveRecord[] = history.map((entry, i) => ({
playerId: publicUserId(entry.playerId),
moveIndex: entry.moveIndex,
boardSnapshot: entry.boardSnapshot,
gameSteps: entry.gameSteps,
at: game.moveTimes[i] ?? endedAt,
}));
const record: GameRecord = {
id: game.id,
player1: playerSummary(mancalaGame.player1Id),
player2: playerSummary(mancalaGame.player2Id),
board: mancalaGame.board.getStoneArray(),
result: getResult(game, ending),
endReason: ending.endReason,
startedAt: game.startedAt,
endedAt,
moveCount: moves.length,
peakSpectators: game.peakSpectators,
totalSpectators: game.allSpectatorIds.length,
moves,
};
if (ending.leftUserId !== undefined) record.leftUserId = publicUserId(ending.leftUserId);
return record;
}

View File

@ -3,6 +3,7 @@ import { BackendGame } from "../../models/Game";
export interface GameStore {
get(id: string): BackendGame | undefined;
getAll(): BackendGame[];
set(id: string, game: BackendGame): void;
remove(id: string): void;

View File

@ -8,6 +8,9 @@ export class GameStoreImpl implements GameStore {
get(id: string): BackendGame | undefined {
return this.gameStore.get(id);
}
getAll(): BackendGame[] {
return [...this.gameStore.values()];
}
set(id: string, game: BackendGame): void {
this.gameStore.set(id, game);
}

View File

@ -0,0 +1,11 @@
import { createHash } from "crypto";
import { KEYCLOAK_USER_ID_PREFIX } from "../auth";
const LEGACY_PUBLIC_ID_PREFIX = "legacy:";
// Ids of legacy userKey users are the secret they log in with, so anything saved or listed publicly uses a hash
// instead. Keycloak ids are not secret and stay as they are.
export function publicUserId(userId: string): string {
if (userId.startsWith(KEYCLOAK_USER_ID_PREFIX) || userId.startsWith(LEGACY_PUBLIC_ID_PREFIX)) return userId;
return LEGACY_PUBLIC_ID_PREFIX + createHash("sha256").update(userId).digest("hex").slice(0, 20);
}

View File

@ -6,20 +6,71 @@ import { MatchMakerImpl } from "./matchmaker/MatchMakerImpl";
import { ExpressApp } from "./server/ExpressApp";
import { WebServer } from "./server/WebServer";
import { KeycloakTokenVerifier, readAuthConfig } from "./auth";
import { createPool, runMigrations } from "./db/postgres";
import { PostgresGameHistoryRepository } from "./history/GameHistoryRepository";
import { createValkeyClient } from "./live/valkey";
import { ValkeyLiveGameStore } from "./live/LiveGameStore";
import { GameRecorder } from "./game/GameRecorder";
const authConfig = readAuthConfig();
const tokenVerifier = new KeycloakTokenVerifier({
const DATABASE_URL = process.env.DATABASE_URL || "postgres://mancala:mancala@localhost:5433/mancala";
const VALKEY_URL = process.env.VALKEY_URL || "redis://localhost:6379";
const MIGRATION_RETRY_MS = 30_000;
const VALKEY_STARTUP_WAIT_MS = 5_000;
async function main() {
const authConfig = readAuthConfig();
const tokenVerifier = new KeycloakTokenVerifier({
issuer: authConfig.keycloakIssuer,
jwksUrl: authConfig.keycloakJwksUrl,
allowedClientIds: authConfig.keycloakClientIds,
});
const pool = createPool(DATABASE_URL);
migrateWhenReachable(pool);
const valkey = createValkeyClient(VALKEY_URL);
await waitForReady(valkey, VALKEY_STARTUP_WAIT_MS);
const liveGameStore = new ValkeyLiveGameStore(valkey);
const recorder = new GameRecorder(new PostgresGameHistoryRepository(pool), liveGameStore);
const rtmt = new RTMTWS({ tokenVerifier, legacyUserKeyAuth: authConfig.legacyUserKeyAuth });
const gameStore = new GameStoreImpl();
const matchMaker = new MatchMakerImpl({ rtmt });
const gameManager = new GameManager({ rtmt, gameStore, matchMaker, liveGameStore, recorder });
// Before accepting connections, so reconnecting players find their games.
await gameManager.restoreLiveGames();
recorder.start();
gameManager.startTimers();
const expressApp = new ExpressApp({ gameStore, legacyUserKeyAuth: authConfig.legacyUserKeyAuth });
const server = new WebServer({ expressApp: expressApp.app });
rtmt.initWebSocket(server.server);
}
// The game runs without the database; saved games wait in the recorder until migrations have run.
function migrateWhenReachable(pool: ReturnType<typeof createPool>) {
runMigrations(pool).catch((err) => {
console.error(`(DB) Migrations failed, retrying in ${MIGRATION_RETRY_MS / 1000}s:`, err.message);
setTimeout(() => migrateWhenReachable(pool), MIGRATION_RETRY_MS);
});
}
function waitForReady(client: ReturnType<typeof createValkeyClient>, timeoutMs: number): Promise<void> {
if (client.status === "ready") return Promise.resolve();
return new Promise((resolve) => {
const timer = setTimeout(() => {
console.error("(Valkey) Not reachable yet; starting without restoring live games");
resolve();
}, timeoutMs);
client.once("ready", () => {
clearTimeout(timer);
resolve();
});
});
}
main().catch((err) => {
console.error(err);
process.exit(1);
});
const rtmt = new RTMTWS({ tokenVerifier, legacyUserKeyAuth: authConfig.legacyUserKeyAuth });
const gameStore = new GameStoreImpl();
const matchMaker = new MatchMakerImpl({ rtmt });
const gameManager = new GameManager({ rtmt, gameStore, matchMaker })
const expressApp = new ExpressApp({ gameStore, legacyUserKeyAuth: authConfig.legacyUserKeyAuth });
const server = new WebServer({ expressApp: expressApp.app });
rtmt.initWebSocket(server.server);

View File

@ -8,7 +8,14 @@ export interface PlayerInfo {
}
export interface BackendGame extends Game {
// Users watching right now.
spectatorIds: string[];
// Everyone who watched at some point, and the most at once.
allSpectatorIds: string[];
peakSpectators: number;
// Keyed by user id, captured when the game is created.
players: Record<string, PlayerInfo>;
startedAt: number;
// Time of each move, aligned with mancalaGame.history.
moveTimes: number[];
}

View File

@ -3,72 +3,347 @@ import {
channel_leave_game,
channel_listen_game_events,
channel_new_game,
channel_on_game_end,
channel_on_game_start,
channel_on_game_update,
channel_on_game_user_leave,
channel_on_new_game_rejected,
channel_on_user_connection_change,
INACTIVE_TIMEOUT_MS,
TURN_TIMEOUT_MS,
} from "@mancala/core";
import { GameManager } from "../../src/game/GameManager";
import { GameRecorder } from "../../src/game/GameRecorder";
import { GameStoreImpl } from "../../src/game/gamestore/GameStoreImpl";
import { publicUserId } from "../../src/game/publicIds";
import { MatchMakerImpl } from "../../src/matchmaker/MatchMakerImpl";
import { BackendGame } from "../../src/models/Game";
import { FakeRtmt } from "./FakeRtmt";
import { flushPromises, InMemoryGameHistoryRepository, InMemoryLiveGameStore } from "./fakes";
const ALICE = { id: "kc:alice", name: "alice", isGuest: false };
const BOB = { id: "kc:bob", name: "bob", isGuest: false };
const BOB = { id: "kc:bob", name: "guest-1a2b3c4d", isGuest: true };
const CAROL = { id: "kc:carol", name: "carol", isGuest: false };
const DAVE = { id: "kc:dave", name: "dave", isGuest: false };
const LEGACY = { id: "0b9f6d1e-7c1a-4f7e-9a51-3f0c2b6d8e11", isGuest: true };
describe("GameManager", () => {
let now: number;
let rtmt: FakeRtmt;
let gameStore: GameStoreImpl;
let liveGameStore: InMemoryLiveGameStore;
let repository: InMemoryGameHistoryRepository;
let gameManager: GameManager;
beforeEach(() => {
const createManager = () => {
rtmt = new FakeRtmt();
gameStore = new GameStoreImpl();
new GameManager({ rtmt, gameStore, matchMaker: new MatchMakerImpl({ rtmt }) });
[ALICE, BOB, CAROL].forEach((user) => rtmt.connect(user));
const recorder = new GameRecorder(repository, liveGameStore);
gameManager = new GameManager({
rtmt, gameStore, matchMaker: new MatchMakerImpl({ rtmt }), liveGameStore, recorder, clock: () => now,
});
return recorder;
};
beforeEach(() => {
now = Date.UTC(2026, 9, 16, 12);
liveGameStore = new InMemoryLiveGameStore();
repository = new InMemoryGameHistoryRepository();
createManager();
[ALICE, BOB, CAROL, DAVE].forEach((user) => rtmt.connect(user));
});
const startGame = (): BackendGame => {
rtmt.receive(ALICE.id, channel_new_game);
rtmt.receive(BOB.id, channel_new_game);
const game = gameStore.getGameByUser(ALICE.id);
const startGame = (first = ALICE, second = BOB): BackendGame => {
rtmt.receive(first.id, channel_new_game);
rtmt.receive(second.id, channel_new_game);
const game = gameStore.getGameByUser(first.id);
if (!game) throw new Error("game did not start");
return game;
};
// Plays the current player's first non-empty pit.
const playAnyMove = (game: BackendGame) => {
const { turnPlayerId } = game.mancalaGame;
const pit = [0, 1, 2, 3, 4, 5].find((p) =>
game.mancalaGame.board.pits[game.mancalaGame.getBoardIndexByPlayerId(turnPlayerId, p)].stoneCount > 0)!;
rtmt.receive(turnPlayerId, channel_game_move, { index: pit });
};
// Leaves the given player a single stone next to their store; playing it empties their side and ends the game.
const setUpLastMove = (game: BackendGame, playerId: string) => {
const isPlayer1 = game.mancalaGame.player1Id === playerId;
const pits = game.mancalaGame.board.pits;
pits.forEach((pit) => (pit.stoneCount = 0));
pits[isPlayer1 ? 5 : 12].stoneCount = 1;
pits[isPlayer1 ? 7 : 0].stoneCount = 1;
game.mancalaGame.turnPlayerId = playerId;
};
describe("while a game is played", () => {
it("sends the connection info once when a game starts", () => {
startGame();
expect(rtmt.messagesTo(ALICE.id, channel_on_game_start)).toHaveLength(1);
expect(rtmt.messagesTo(ALICE.id, channel_on_user_connection_change)).toHaveLength(1);
});
it("removes a game from memory when a player leaves", () => {
const game = startGame();
rtmt.receive(ALICE.id, channel_leave_game);
expect(gameStore.get(game.id)).toBeUndefined();
expect(gameStore.getGameByUser(BOB.id)).toBeUndefined();
});
it("removes a game from memory when it ends by the rules", () => {
const game = startGame();
// Alice has a single stone in her last pit, next to her store; playing it empties her side.
const aliceIsPlayer1 = game.mancalaGame.player1Id === ALICE.id;
const board = game.mancalaGame.board;
board.pits.forEach((pit) => (pit.stoneCount = 0));
board.pits[aliceIsPlayer1 ? 5 : 12].stoneCount = 1;
board.pits[aliceIsPlayer1 ? 7 : 0].stoneCount = 1;
game.mancalaGame.turnPlayerId = ALICE.id;
rtmt.receive(ALICE.id, channel_game_move, { index: 5 });
expect(game.mancalaGame.state).toBe("ended");
expect(gameStore.get(game.id)).toBeUndefined();
});
it("stops counting a spectator who disconnects", () => {
it("does not send spectator ids to clients", () => {
const game = startGame();
rtmt.receive(CAROL.id, channel_listen_game_events, game.id);
expect(game.spectatorIds).toEqual([CAROL.id]);
expect(gameStore.getSpeactatingGameIdsByUser(CAROL.id)).toEqual([game.id]);
playAnyMove(game);
const update = rtmt.messagesTo(ALICE.id, channel_on_game_update).pop();
expect(update).not.toHaveProperty("spectatorIds");
expect(update).not.toHaveProperty("allSpectatorIds");
expect(update.spectatorCount).toBe(1);
});
it("records the time of each move", () => {
const game = startGame();
now += 1000;
playAnyMove(game);
now += 2000;
playAnyMove(game);
expect(game.moveTimes).toEqual([game.startedAt + 1000, game.startedAt + 3000]);
expect(rtmt.messagesTo(ALICE.id, channel_on_game_update).pop().moveTimes).toHaveLength(2);
});
it("rejects a new game for a player who has one in progress", () => {
const game = startGame();
rtmt.receive(ALICE.id, channel_new_game);
rtmt.receive(CAROL.id, channel_new_game);
expect(rtmt.messagesTo(ALICE.id, channel_on_new_game_rejected)).toEqual([{ gameId: game.id }]);
expect(gameStore.getGameByUser(ALICE.id)).toBe(game);
expect(gameStore.getGameByUser(CAROL.id)).toBeUndefined();
});
it("counts spectators: removed on disconnect, peak and distinct total kept", async () => {
const game = startGame();
rtmt.receive(CAROL.id, channel_listen_game_events, game.id);
rtmt.receive(DAVE.id, channel_listen_game_events, game.id);
rtmt.disconnect(CAROL.id);
expect(game.spectatorIds).toEqual([]);
rtmt.connect(CAROL);
rtmt.receive(CAROL.id, channel_listen_game_events, game.id);
rtmt.disconnect(DAVE.id);
expect(game.spectatorIds).toEqual([CAROL.id]);
expect(game.peakSpectators).toBe(2);
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
expect(repository.records.get(game.id)).toMatchObject({ peakSpectators: 2, totalSpectators: 2 });
});
it("keeps a snapshot of the live game until it ends", async () => {
const game = startGame();
playAnyMove(game);
await flushPromises();
expect(liveGameStore.games.get(game.id)).toMatchObject({ id: game.id, moveTimes: [now] });
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
expect(liveGameStore.games.has(game.id)).toBe(false);
});
});
describe("how games end", () => {
it("completed: the winner by the rules, with every move saved", async () => {
const game = startGame();
playAnyMove(game);
setUpLastMove(game, ALICE.id);
rtmt.receive(ALICE.id, channel_game_move, { index: 5 });
await flushPromises();
// Alice's stone goes into her store and empties her side, so she also collects Bob's last stone: 2-0.
const aliceSeat = game.mancalaGame.player1Id === ALICE.id ? "player1" : "player2";
const record = repository.records.get(game.id)!;
expect(record.endReason).toBe("completed");
expect(record.result).toBe(aliceSeat);
expect(record.moves).toHaveLength(2);
expect(record.moves.map((m) => m.at)).toEqual(game.moveTimes);
expect(gameStore.get(game.id)).toBeUndefined();
expect(rtmt.messagesTo(BOB.id, channel_on_game_end)).toEqual([
{ gameId: game.id, result: aliceSeat, endReason: "completed", winnerId: ALICE.id },
]);
});
it("left: the player who leaves loses", async () => {
const game = startGame();
rtmt.receive(CAROL.id, channel_listen_game_events, game.id);
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
const record = repository.records.get(game.id)!;
const bobSeat = game.mancalaGame.player1Id === BOB.id ? "player1" : "player2";
expect(record).toMatchObject({ endReason: "left", result: bobSeat, leftUserId: ALICE.id });
for (const user of [ALICE, BOB, CAROL]) {
expect(rtmt.messagesTo(user.id, channel_on_game_user_leave)).toEqual([ALICE.id]);
expect(rtmt.messagesTo(user.id, channel_on_game_end)).toEqual([
{ gameId: game.id, result: bobSeat, endReason: "left", winnerId: BOB.id, leftUserId: ALICE.id },
]);
}
});
it("timeout: the player whose turn it is loses after 5 minutes offline", async () => {
const game = startGame();
const turnPlayer = game.mancalaGame.turnPlayerId;
const waitingPlayer = turnPlayer === ALICE.id ? BOB.id : ALICE.id;
rtmt.disconnect(turnPlayer);
const offlineAt = now;
const info = rtmt.messagesTo(waitingPlayer, channel_on_user_connection_change).pop();
const turnPlayerInfo = [info.user1ConnectionInfo, info.user2ConnectionInfo].find((i: any) => i.userId === turnPlayer);
expect(turnPlayerInfo).toMatchObject({ isOnline: false, offlineSince: offlineAt });
now = offlineAt + TURN_TIMEOUT_MS - 1;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
now = offlineAt + TURN_TIMEOUT_MS;
gameManager.checkTimeouts();
await flushPromises();
expect(gameStore.get(game.id)).toBeUndefined();
const record = repository.records.get(game.id)!;
expect(record.endReason).toBe("timeout");
expect(record.result).toBe(game.mancalaGame.player1Id === waitingPlayer ? "player1" : "player2");
});
it("timeout does not apply when the waiting player is offline too, or when it is not the offline player's turn", () => {
const game = startGame();
const turnPlayer = game.mancalaGame.turnPlayerId;
const waitingPlayer = turnPlayer === ALICE.id ? BOB.id : ALICE.id;
rtmt.disconnect(waitingPlayer);
now += TURN_TIMEOUT_MS * 2;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
rtmt.disconnect(turnPlayer);
now += TURN_TIMEOUT_MS * 2;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
});
it("a reconnect resets the offline time", () => {
const game = startGame();
const turnPlayer = game.mancalaGame.turnPlayerId;
const user = turnPlayer === ALICE.id ? ALICE : BOB;
rtmt.disconnect(turnPlayer);
now += TURN_TIMEOUT_MS - 1000;
rtmt.connect(user);
rtmt.disconnect(turnPlayer);
now += 2000;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
});
it("inactive: both players offline for 30 minutes ends the game without a winner", async () => {
const game = startGame();
rtmt.disconnect(ALICE.id);
now += 60_000;
rtmt.disconnect(BOB.id);
now += INACTIVE_TIMEOUT_MS - 1;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
now += 1;
gameManager.checkTimeouts();
await flushPromises();
expect(repository.records.get(game.id)).toMatchObject({ endReason: "inactive", result: null });
});
});
describe("saving", () => {
it("replaces legacy ids with public ids in saved games", async () => {
rtmt.connect({ id: LEGACY.id, isGuest: true });
const game = startGame(ALICE, { ...LEGACY, name: undefined } as any);
playAnyMove(game);
rtmt.receive(LEGACY.id, channel_leave_game);
await flushPromises();
const saved = JSON.stringify(repository.records.get(game.id));
expect(saved).not.toContain(LEGACY.id);
expect(saved).toContain(publicUserId(LEGACY.id));
expect(saved).toContain(ALICE.id);
});
it("keeps a game that could not be saved and saves it once the database is back", async () => {
repository.failing = true;
const game = startGame();
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
expect(liveGameStore.pending.has(game.id)).toBe(true);
// A restarted backend picks the pending record up from Valkey.
repository.failing = false;
const recorder = createManager();
await recorder.retryPending();
expect(repository.records.has(game.id)).toBe(true);
expect(liveGameStore.pending.has(game.id)).toBe(false);
});
it("keeps games playable while Valkey and the database are down", async () => {
liveGameStore.failing = true;
repository.failing = true;
const game = startGame();
playAnyMove(game);
playAnyMove(game);
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
expect(game.mancalaGame.history).toHaveLength(2);
expect(gameStore.get(game.id)).toBeUndefined();
});
});
describe("restart", () => {
it("restores live games, which can then be played on", async () => {
const game = startGame();
playAnyMove(game);
playAnyMove(game);
await flushPromises();
const board = game.mancalaGame.board.getStoneArray();
const turnPlayer = game.mancalaGame.turnPlayerId;
createManager();
expect(await gameManager.restoreLiveGames()).toBe(1);
const restored = gameStore.getGameByUser(ALICE.id)!;
expect(restored.id).toBe(game.id);
expect(restored.mancalaGame.board.getStoneArray()).toEqual(board);
expect(restored.mancalaGame.turnPlayerId).toBe(turnPlayer);
expect(restored.mancalaGame.history).toHaveLength(2);
expect(gameStore.getGameByUser(BOB.id)).toBe(restored);
rtmt.connect(ALICE);
rtmt.connect(BOB);
expect(rtmt.messagesTo(ALICE.id, channel_on_game_update)).toHaveLength(1);
playAnyMove(restored);
expect(restored.mancalaGame.history).toHaveLength(3);
});
it("counts players as offline from the restart, so the timeout rules start over", () => {
startGame();
createManager();
return gameManager.restoreLiveGames().then(() => {
const game = gameStore.getGameByUser(ALICE.id)!;
const waiting = game.mancalaGame.turnPlayerId === ALICE.id ? BOB : ALICE;
rtmt.connect(waiting);
now += TURN_TIMEOUT_MS - 1;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBe(game);
now += 1;
gameManager.checkTimeouts();
expect(gameStore.get(game.id)).toBeUndefined();
});
});
it("does not bring back a game that has already been saved", async () => {
const game = startGame();
await flushPromises();
const snapshot = liveGameStore.games.get(game.id);
rtmt.receive(ALICE.id, channel_leave_game);
await flushPromises();
liveGameStore.games.set(game.id, snapshot); // as if removing it had failed
createManager();
expect(await gameManager.restoreLiveGames()).toBe(0);
await flushPromises();
expect(liveGameStore.games.has(game.id)).toBe(false);
});
});
});

View File

@ -0,0 +1,65 @@
import { FinishedGameSummary, GameRecord } from "@mancala/core";
import { GameHistoryRepository } from "../../src/history/GameHistoryRepository";
import { LiveGameStore } from "../../src/live/LiveGameStore";
export class InMemoryLiveGameStore implements LiveGameStore {
readonly games = new Map<string, unknown>();
readonly pending = new Map<string, GameRecord>();
failing = false;
private check() {
if (this.failing) return Promise.reject(new Error("Valkey is down"));
return Promise.resolve();
}
async saveGame(gameId: string, snapshot: unknown) {
await this.check();
this.games.set(gameId, JSON.parse(JSON.stringify(snapshot)));
}
async removeGame(gameId: string) {
await this.check();
this.games.delete(gameId);
}
async loadGames() {
await this.check();
return [...this.games.values()];
}
async savePendingRecord(record: GameRecord) {
await this.check();
this.pending.set(record.id, record);
}
async removePendingRecord(gameId: string) {
await this.check();
this.pending.delete(gameId);
}
async loadPendingRecords() {
await this.check();
return [...this.pending.values()];
}
}
export class InMemoryGameHistoryRepository implements GameHistoryRepository {
readonly records = new Map<string, GameRecord>();
failing = false;
async save(record: GameRecord) {
if (this.failing) throw new Error("database is down");
if (!this.records.has(record.id)) this.records.set(record.id, record);
}
async get(id: string) {
if (this.failing) throw new Error("database is down");
return this.records.get(id);
}
async getRecentForUser(userId: string, limit: number): Promise<FinishedGameSummary[]> {
return [...this.records.values()]
.filter((r) => r.player1.userId === userId || r.player2.userId === userId)
.sort((a, b) => b.endedAt - a.endedAt).slice(0, limit);
}
async getPopular(limit: number): Promise<FinishedGameSummary[]> {
return [...this.records.values()]
.sort((a, b) => b.peakSpectators - a.peakSpectators || b.endedAt - a.endedAt).slice(0, limit);
}
}
// Lets fire-and-forget promises (saving to Valkey or the database) settle.
export const flushPromises = () => new Promise((resolve) => setImmediate(resolve));

View File

@ -119,7 +119,11 @@ src/game/publicIds.ts publicUserId and helpers
- **New game while playing** (R2.3): `MatchMaker` is not called; the user gets `on_new_game_rejected` with the game
id.
- **Restart** (R1.2): on startup all snapshots are loaded from Valkey and put back into memory; both players count as
offline from that moment, so the timeout rules start fresh.
offline from that moment, so the timeout rules start fresh. Snapshots of games that are already saved are dropped instead. If Valkey is
unreachable at startup, the backend starts without restoring.
- **Messages**: players and spectators get a trimmed game (`id`, `mancalaGame`, `gameUsersConnectionInfo`,
`startedAt`, `moveTimes`, `spectatorCount`), never the spectator ids. When a game ends for any reason they also get
`on_game_end` (`GameEndMessage`); a player leaving still sends `on_game_user_leave` as before.
- **Fixes needed by this feature**: spectators are removed when they disconnect (the `for…in` loop in
`GameStoreImpl`), finished games are removed from memory, and the connection info is sent once at game start.

View File

@ -12,7 +12,7 @@ Work through the tasks in order; each ends with its tests passing and a commit.
- [x] 4. **Postgres repository**: pool, migration runner, `games` and `game_players`, `PostgresGameHistoryRepository`
(save, get, recent for user, popular), integration tests.
- [x] 5. **Valkey store**: live game snapshots and restore, pending saves, cache; integration tests.
- [ ] 6. **Game lifecycle**: start and move times, spectator counters, outcomes (completed, left, timeout, inactive),
- [x] 6. **Game lifecycle**: start and move times, spectator counters, outcomes (completed, left, timeout, inactive),
timeout watcher, `offlineSince`, rejecting a new game while playing, saving with retry, restore on startup;
unit tests.
- [ ] 7. **API**: public ids, Bearer auth middleware, `/games/me/active`, `/games/me`, `/games/popular`,

View File

@ -64,3 +64,12 @@ export interface FinishedGameResponse {
status: "finished";
record: GameRecord;
}
// on_game_end, sent to the players and spectators of a live game when it ends for any reason.
export interface GameEndMessage {
gameId: string;
result: GameResult | null;
endReason: EndReason;
winnerId?: string;
leftUserId?: string;
}