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

package app.playmancala.game;

import java.util.Comparator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import app.playmancala.loop.Cancellable;
import app.playmancala.loop.Scheduler;
import app.playmancala.model.Channels;
import app.playmancala.model.Json;
import app.playmancala.model.LiveGameSummary;
import app.playmancala.rtmt.Rtmt;

// Live games for the home page: most spectated first, then most recently active. Subscribers get the top games
// when they subscribe and again whenever they change, at most once per second. Subscribing does not make anyone a
// spectator. Only touched on the game loop.
public class LiveGames {
    public static final int LIVE_GAMES_PUSH_COUNT = 5;
    private static final long PUSH_INTERVAL_MS = 1_000;

    private final Rtmt rtmt;
    private final GameStore gameStore;
    private final Scheduler scheduler;
    private final Set<String> subscribers = new LinkedHashSet<>();
    private String lastPushed = "";
    private long lastPushAt;
    private Cancellable pushTimer;

    public LiveGames(Rtmt rtmt, GameStore gameStore, GameManager gameManager, Scheduler scheduler) {
        this.rtmt = rtmt;
        this.gameStore = gameStore;
        this.scheduler = scheduler;
        rtmt.listenMessage(Channels.SUBSCRIBE_LIVE_GAMES, (userId, message) -> subscribe(userId));
        rtmt.listenMessage(Channels.UNSUBSCRIBE_LIVE_GAMES, (userId, message) -> subscribers.remove(userId));
        rtmt.listenOnClientConnectionChange((userId, isOnline) -> {
            if (!isOnline) subscribers.remove(userId);
        });
        gameManager.onLiveGamesChanged(this::schedulePush);
    }

    public static LiveGameSummary toLiveGameSummary(BackendGame game) {
        var mancalaGame = game.mancalaGame;
        return new LiveGameSummary(game.id,
                GameSerialization.publicSummary(game, mancalaGame.getPlayer1Id()),
                GameSerialization.publicSummary(game, mancalaGame.getPlayer2Id()),
                GameSerialization.boxed(mancalaGame.getBoard().getStoneArray()),
                PublicIds.publicUserId(mancalaGame.getTurnPlayerId()), game.spectatorIds.size(),
                mancalaGame.getHistory().size(), game.startedAt);
    }

    private static long lastActivity(BackendGame game) {
        return game.moveTimes.isEmpty() ? game.startedAt : game.moveTimes.getLast();
    }

    public List<LiveGameSummary> getTop(int limit) {
        return gameStore.getAll().stream()
                .sorted(Comparator.<BackendGame>comparingInt(game -> game.spectatorIds.size())
                        .thenComparingLong(LiveGames::lastActivity).reversed())
                .limit(limit)
                .map(LiveGames::toLiveGameSummary)
                .toList();
    }

    public void dispose() {
        if (pushTimer != null) pushTimer.cancel();
    }

    private void subscribe(String userId) {
        subscribers.add(userId);
        rtmt.sendMessage(userId, Channels.ON_LIVE_GAMES, Map.of("games", getTop(LIVE_GAMES_PUSH_COUNT)));
    }

    private void schedulePush() {
        if (pushTimer != null) return;
        long wait = Math.max(0, lastPushAt + PUSH_INTERVAL_MS - scheduler.now());
        pushTimer = scheduler.schedule(() -> {
            pushTimer = null;
            push();
        }, wait);
    }

    private void push() {
        List<LiveGameSummary> games = getTop(LIVE_GAMES_PUSH_COUNT);
        String serialized = Json.write(games);
        if (serialized.equals(lastPushed)) return;
        lastPushed = serialized;
        lastPushAt = scheduler.now();
        Map<String, Object> message = Map.of("games", games);
        List.copyOf(subscribers).forEach(userId -> rtmt.sendMessage(userId, Channels.ON_LIVE_GAMES, message));
    }
}