package app.playmancala.game; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.Executor; import java.util.concurrent.ThreadLocalRandom; import java.util.function.LongSupplier; import org.jboss.logging.Logger; import com.fasterxml.jackson.databind.JsonNode; import app.playmancala.engine.CommonMancalaGame; import app.playmancala.engine.GameState; import app.playmancala.engine.MancalaGame; import app.playmancala.game.GameSerialization.GameEnding; import app.playmancala.game.GameSerialization.LiveGameSnapshot; import app.playmancala.live.LiveGameStore; import app.playmancala.model.AuthUser; import app.playmancala.model.Channels; import app.playmancala.model.ClientGame; import app.playmancala.model.EndReason; import app.playmancala.model.GameEndMessage; import app.playmancala.model.GameRecord; import app.playmancala.model.GameUsersConnectionInfo; import app.playmancala.model.PlayerSummary; import app.playmancala.model.Seat; import app.playmancala.model.UserConnectionInfo; import app.playmancala.rtmt.Rtmt; // Runs live games: matchmaking, moves, the clock, spectators and every way a game ends. Only touched on the game loop. // Writes to Valkey go through `storeExecutor`, saving finished games through the recorder; neither blocks the loop. public class GameManager { private static final Logger LOG = Logger.getLogger(GameManager.class); private final GameStore gameStore; private final Rtmt rtmt; private final MatchMaker matchMaker; private final LiveGameStore liveGameStore; private final GameRecorder recorder; private final Executor storeExecutor; private final LongSupplier clock; // Time each player has for a new game. private final long gameClockMs; private final List liveGamesListeners = new ArrayList<>(); public GameManager(GameStore gameStore, Rtmt rtmt, MatchMaker matchMaker, LiveGameStore liveGameStore, GameRecorder recorder, Executor storeExecutor, LongSupplier clock, long gameClockMs) { this.gameStore = gameStore; this.rtmt = rtmt; this.matchMaker = matchMaker; this.liveGameStore = liveGameStore; this.recorder = recorder; this.storeExecutor = storeExecutor; this.clock = clock; this.gameClockMs = gameClockMs; matchMaker.listenOnPlayersPaired((player1Id, player2Id) -> startGame(createMancalaGame(player1Id, player2Id))); listenRtmtMessages(); listenUserConnectionChange(); } // Called whenever a live game starts, changes or ends. public void onLiveGamesChanged(Runnable listener) { liveGamesListeners.add(listener); } private void listenRtmtMessages() { rtmt.listenMessage(Channels.NEW_GAME, (userId, message) -> { BackendGame activeGame = gameStore.getGameByUser(userId); if (activeGame != null) { rtmt.sendMessage(userId, Channels.ON_NEW_GAME_REJECTED, Map.of("gameId", activeGame.id)); return; } matchMaker.join(userId); }); rtmt.listenMessage(Channels.CANCEL_NEW_GAME, (userId, message) -> matchMaker.cancel(userId)); rtmt.listenMessage(Channels.GAME_MOVE, this::onGameMove); rtmt.listenMessage(Channels.LEAVE_GAME, (userId, message) -> onPlayerLeave(userId)); rtmt.listenMessage(Channels.LISTEN_GAME_EVENTS, (userId, message) -> { if (message.isTextual()) addSpectator(userId, message.asText()); }); rtmt.listenMessage(Channels.UNLISTEN_GAME_EVENTS, (userId, message) -> { if (message.isTextual()) removeSpectator(userId, message.asText()); }); } private void onGameMove(String userId, JsonNode gameMove) { BackendGame game = gameStore.getGameByUser(userId); if (game == null) { LOG.debug("Game not found!"); return; } MancalaGame mancalaGame = game.mancalaGame; try { long now = clock.getAsLong(); // A move that arrives after the mover's time has run out is not played. if (userId.equals(mancalaGame.getTurnPlayerId()) && isOutOfTime(game, now)) { endGameOnTime(game); return; } JsonNode index = gameMove.path("index"); if (!index.canConvertToInt() || !index.isIntegralNumber()) { throw new IllegalStateException("Invalid move: " + gameMove); } int movesBefore = mancalaGame.getHistory().size(); Seat moverSeat = seatOf(mancalaGame, userId); mancalaGame.moveByPlayerPit(userId, index.asInt()); if (mancalaGame.getHistory().size() > movesBefore && moverSeat != null) { game.moveTimes.add(now); GameClocks.ChargedMove charged = GameClocks.chargeMove(game.clock, moverSeat, now); game.clock = charged.clock(); game.moveRemainingMs.add(charged.remainingMs()); } game.gameUsersConnectionInfo = getGameUsersConnectionInfo(game); sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_UPDATE, toClientGame(game)); if (mancalaGame.getState() == GameState.ENDED) { endGame(game, new GameEnding(EndReason.COMPLETED, mancalaGame.getWonPlayerId(), null)); } else { saveLiveGame(game); notifyLiveGamesChanged(); } } catch (RuntimeException e) { onGameError(game, e); } } private void onGameError(BackendGame game, RuntimeException error) { LOG.error("(GameManager) Game " + game.id + " failed", error); String crash = CrashLog.describe(error, game); String fileName = CrashLog.fileName(game.id); storeExecutor.execute(() -> CrashLog.write(fileName, crash)); LOG.infof("Game crash saved to file : %s", fileName); // The Node backend sent the Error object, which serialises to {}; clients only use the channel. sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_CRASHED, Map.of()); } private void onPlayerLeave(String userId) { BackendGame game = gameStore.getGameByUser(userId); if (game != null) { endGame(game, new GameEnding(EndReason.LEFT, game.opponentOf(userId), userId)); } } // Ends a live game for any reason: stops the clock, tells everyone, forgets the game and saves it. private void endGame(BackendGame game, GameEnding ending) { if (gameStore.get(game.id) != game) return; deleteGame(game); game.clock = GameClocks.stop(game.clock, getRunningSeat(game), clock.getAsLong()); GameEndMessage endMessage = new GameEndMessage(game.id, GameSerialization.getResult(game, ending), ending.endReason(), game.clock, ending.winnerId(), ending.leftUserId()); if (ending.leftUserId() != null) { sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_USER_LEAVE, ending.leftUserId()); } sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_END, endMessage); game.spectatorIds.clear(); GameRecord record = GameSerialization.toGameRecord(game, ending, clock.getAsLong()); inStore(() -> liveGameStore.removeGame(game.id)); recorder.save(record); notifyLiveGamesChanged(); } // The player whose turn it is ran out of time: the opponent wins. private void endGameOnTime(BackendGame game) { endGame(game, new GameEnding(EndReason.TIMEOUT, game.opponentOf(game.mancalaGame.getTurnPlayerId()), null)); } private boolean isOutOfTime(BackendGame game, long now) { Seat runningSeat = getRunningSeat(game); return runningSeat != null && GameClocks.getRemainingMs(game.clock, runningSeat, runningSeat, now) <= 0; } // The seat whose clock runs; null once the game has ended. private Seat getRunningSeat(BackendGame game) { MancalaGame mancalaGame = game.mancalaGame; if (mancalaGame.getState() == GameState.ENDED || game.clock.turnStartedAt() == null) return null; return seatOf(mancalaGame, mancalaGame.getTurnPlayerId()); } private static Seat seatOf(MancalaGame mancalaGame, String playerId) { return GameClocks.getSeat(mancalaGame.getPlayer1Id(), mancalaGame.getPlayer2Id(), playerId); } // Ends every live game whose running clock has reached zero; run by GameClockScheduler. The clock runs whether or // not the players are online, so this also ends games that both players have left. public void endGamesOutOfTime() { long now = clock.getAsLong(); for (BackendGame game : gameStore.getAll()) { try { if (isOutOfTime(game, now)) endGameOnTime(game); } catch (RuntimeException e) { LOG.error("(GameManager) Could not end game " + game.id + " on time", e); } } } // Puts the live games saved in Valkey back into memory, e.g. after a restart or deploy. Blocks on Valkey and the // database; runs once at startup. public int restoreLiveGames() { List stored; try { stored = liveGameStore.loadGames(); } catch (RuntimeException e) { LOG.errorf("(GameManager) Could not load live games: %s", e.getMessage()); return 0; } int restored = 0; for (JsonNode json : stored) { LiveGameSnapshot snapshot = GameSerialization.parseSnapshot(json); if (snapshot == null || gameStore.get(snapshot.id()) != null) continue; // A game that ended but whose snapshot could not be removed must not come back. if (isRecorded(snapshot.id())) { inStore(() -> liveGameStore.removeGame(snapshot.id())); continue; } BackendGame game = GameSerialization.fromSnapshot(snapshot, clock.getAsLong(), gameClockMs); gameStore.set(game.id, game); gameStore.setGameByUser(game.mancalaGame.getPlayer1Id(), game); gameStore.setGameByUser(game.mancalaGame.getPlayer2Id(), game); game.gameUsersConnectionInfo = getGameUsersConnectionInfo(game); restored++; } if (restored > 0) { LOG.infof("(GameManager) Restored %d live game(s)", restored); notifyLiveGamesChanged(); } return restored; } private boolean isRecorded(String gameId) { try { return recorder.isRecorded(gameId); } catch (RuntimeException e) { return false; } } private void addSpectator(String userId, String gameId) { BackendGame game = gameStore.get(gameId); if (game == null || game.isPlayer(userId) || game.spectatorIds.contains(userId)) return; game.spectatorIds.add(userId); if (!game.allSpectatorIds.contains(userId)) game.allSpectatorIds.add(userId); game.peakSpectators = Math.max(game.peakSpectators, game.spectatorIds.size()); saveLiveGame(game); notifyLiveGamesChanged(); } private void removeSpectator(String userId, String gameId) { BackendGame game = gameStore.get(gameId); if (game != null && game.spectatorIds.remove(userId)) { notifyLiveGamesChanged(); } } private void sendMessageToPlayersAndSpectators(BackendGame game, String channel, Object message) { rtmt.sendMessage(game.mancalaGame.getPlayer1Id(), channel, message); rtmt.sendMessage(game.mancalaGame.getPlayer2Id(), channel, message); List.copyOf(game.spectatorIds).forEach(spectatorId -> rtmt.sendMessage(spectatorId, channel, message)); } public ClientGame toClientGame(BackendGame game) { return new ClientGame(game.id, game.mancalaGame, game.gameUsersConnectionInfo, game.startedAt, List.copyOf(game.moveTimes), game.spectatorIds.size(), game.clock, clock.getAsLong()); } public BackendGame createMancalaGame(String userId1, String userId2) { MancalaGame mancalaGame = new CommonMancalaGame(UUID.randomUUID().toString(), userId1, userId2); mancalaGame.setTurnPlayerId(ThreadLocalRandom.current().nextBoolean() ? userId1 : userId2); long now = clock.getAsLong(); PlayerSummary player1 = getPlayerInfo(userId1); PlayerSummary player2 = getPlayerInfo(userId2); Map players = new LinkedHashMap<>(); players.put(userId1, player1); players.put(userId2, player2); BackendGame game = new BackendGame(mancalaGame, getGameUsersConnectionInfo(player1, player2), List.of(), 0, players, now, List.of(), GameClocks.start(gameClockMs, now), List.of()); gameStore.set(game.id, game); gameStore.setGameByUser(userId1, game); gameStore.setGameByUser(userId2, game); return game; } private PlayerSummary getPlayerInfo(String userId) { AuthUser user = rtmt.getUser(userId); return new PlayerSummary(userId, user != null ? user.name() : null, user == null || user.isGuest()); } public GameUsersConnectionInfo getGameUsersConnectionInfo(BackendGame game) { return getGameUsersConnectionInfo(game.player(game.mancalaGame.getPlayer1Id()), game.player(game.mancalaGame.getPlayer2Id())); } private GameUsersConnectionInfo getGameUsersConnectionInfo(PlayerSummary player1, PlayerSummary player2) { return new GameUsersConnectionInfo( UserConnectionInfo.of(player1, rtmt.isClientOnline(player1.userId())), UserConnectionInfo.of(player2, rtmt.isClientOnline(player2.userId()))); } public void startGame(BackendGame game) { sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_START, toClientGame(game)); sendUserConnectionInfo(game); saveLiveGame(game); notifyLiveGamesChanged(); } public void sendUserConnectionInfo(BackendGame game) { game.gameUsersConnectionInfo = getGameUsersConnectionInfo(game); sendMessageToPlayersAndSpectators(game, Channels.ON_USER_CONNECTION_CHANGE, game.gameUsersConnectionInfo); } public void deleteGame(BackendGame game) { gameStore.removeGameByPlayer(game.mancalaGame.getPlayer1Id()); gameStore.removeGameByPlayer(game.mancalaGame.getPlayer2Id()); gameStore.remove(game.id); } private void saveLiveGame(BackendGame game) { LiveGameSnapshot snapshot = GameSerialization.toSnapshot(game); inStore(() -> liveGameStore.saveGame(game.id, snapshot)); } // Runs a Valkey write off the loop; a failure is only logged, the game goes on from memory. private void inStore(Runnable write) { storeExecutor.execute(() -> { try { write.run(); } catch (RuntimeException e) { LOG.errorf("(GameManager) Live game store: %s", e.getMessage()); } }); } private void notifyLiveGamesChanged() { liveGamesListeners.forEach(Runnable::run); } private void listenUserConnectionChange() { rtmt.listenOnClientConnectionChange((clientId, isOnline) -> { BackendGame game = gameStore.getGameByUser(clientId); if (game != null) sendUserConnectionInfo(game); if (!isOnline) { gameStore.getSpectatingGameIdsByUser(clientId).forEach(gameId -> removeSpectator(clientId, gameId)); } else if (game != null) { rtmt.sendMessage(clientId, Channels.ON_GAME_UPDATE, toClientGame(game)); } }); } }