mancala/apps/backend/src/main/java/app/playmancala/game/GameManager.java

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.Consumer;
import java.util.function.IntSupplier;
import java.util.function.LongSupplier;

import org.jboss.logging.Logger;

import com.fasterxml.jackson.databind.JsonNode;

import app.playmancala.bot.BotLevel;
import app.playmancala.bot.BotSettings;
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);
    // Clients show the bot by its id ("Bot · Medium"); the name is for anything that only shows names.
    static final String BOT_NAME = "Bot";

    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<Runnable> liveGamesListeners = new ArrayList<>();
    private final List<Consumer<BackendGame>> turnListeners = 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);
    }

    // Called whenever a player is to move: when a game starts or is restored, and after a move that did not end it,
    // including one that gives an extra turn.
    public void onTurnStarted(Consumer<BackendGame> listener) {
        turnListeners.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.NEW_BOT_GAME, this::onNewBotGame);
        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 onNewBotGame(String userId, JsonNode message) {
        BotLevel level = BotLevel.fromWireName(message.path("level").asText(null));
        if (level == null) {
            LOG.debugf("(GameManager) Unknown bot level: %s", message);
            return;
        }
        BackendGame activeGame = gameStore.getGameByUser(userId);
        if (activeGame != null) {
            rtmt.sendMessage(userId, Channels.ON_NEW_GAME_REJECTED, Map.of("gameId", activeGame.id));
            return;
        }
        matchMaker.cancel(userId);
        BotSettings bot = BotSettings.draw(level, ThreadLocalRandom.current());
        startGame(createMancalaGame(userId, bot.botId(), bot));
    }

    private void onGameMove(String userId, JsonNode gameMove) {
        BackendGame game = gameStore.getGameByUser(userId);
        if (game == null) {
            LOG.debug("Game not found!");
            return;
        }
        playMove(game, userId, () -> {
            JsonNode index = gameMove.path("index");
            if (!index.canConvertToInt() || !index.isIntegralNumber()) {
                throw new IllegalStateException("Invalid move: " + gameMove);
            }
            return index.asInt();
        });
    }

    // Plays a player's pit (0-5) in a live game; the bot's moves come here too.
    public void playMove(BackendGame game, String userId, int pitIndex) {
        playMove(game, userId, () -> pitIndex);
    }

    // The pit is read after the clock check, so an invalid move after the time has run out still ends the game on time.
    private void playMove(BackendGame game, String userId, IntSupplier pitIndex) {
        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;
            }
            int index = pitIndex.getAsInt();
            int movesBefore = mancalaGame.getHistory().size();
            Seat moverSeat = seatOf(mancalaGame, userId);
            mancalaGame.moveByPlayerPit(userId, index);
            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();
                notifyTurnStarted(game);
            }
        } 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<JsonNode> 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();
            gameStore.getAll().forEach(this::notifyTurnStarted);
        }
        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) {
        sendMessageToPlayer(game.mancalaGame.getPlayer1Id(), channel, message);
        sendMessageToPlayer(game.mancalaGame.getPlayer2Id(), channel, message);
        List.copyOf(game.spectatorIds).forEach(spectatorId -> rtmt.sendMessage(spectatorId, channel, message));
    }

    // The bot has no connection to send to.
    private void sendMessageToPlayer(String playerId, String channel, Object message) {
        if (!BotLevel.isBotId(playerId)) rtmt.sendMessage(playerId, 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) {
        return createMancalaGame(userId1, userId2, null);
    }

    // `bot` is set for a game against the bot, which plays as one of the two users.
    public BackendGame createMancalaGame(String userId1, String userId2, BotSettings bot) {
        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<String, PlayerSummary> 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(), bot);
        gameStore.set(game.id, game);
        gameStore.setGameByUser(userId1, game);
        gameStore.setGameByUser(userId2, game);
        return game;
    }

    private PlayerSummary getPlayerInfo(String userId) {
        if (BotLevel.isBotId(userId)) return new PlayerSummary(userId, BOT_NAME, false);
        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, isOnline(player1.userId())),
                UserConnectionInfo.of(player2, isOnline(player2.userId())));
    }

    // The bot is always there.
    private boolean isOnline(String userId) {
        return BotLevel.isBotId(userId) || rtmt.isClientOnline(userId);
    }

    public void startGame(BackendGame game) {
        sendMessageToPlayersAndSpectators(game, Channels.ON_GAME_START, toClientGame(game));
        sendUserConnectionInfo(game);
        saveLiveGame(game);
        notifyLiveGamesChanged();
        notifyTurnStarted(game);
    }

    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 notifyTurnStarted(BackendGame game) {
        turnListeners.forEach(listener -> listener.accept(game));
    }

    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));
            }
        });
    }
}